woc.objects

  1import difflib
  2import os
  3import re
  4import warnings
  5from datetime import datetime, timedelta, timezone
  6from functools import cached_property, lru_cache
  7from logging import getLogger
  8from typing import Dict, Generator, List, Optional, Set, Tuple, Union
  9
 10from .base import WocMapsBase
 11from .local import fnvhash
 12
 13_global_woc: Optional[WocMapsBase] = None
 14_logger = getLogger(__name__)
 15_DAY_Z = datetime.fromtimestamp(0, tz=None)
 16
 17
 18def init_woc_objects(woc: WocMapsBase):
 19    """
 20    Stores wocMaps object globally so you don't have to pass it around.
 21
 22    :param woc: a wocMaps object.
 23    """
 24    global _global_woc
 25    _global_woc = woc
 26
 27
 28@lru_cache(maxsize=None)
 29def parse_timezone_offset(offset_str: str) -> timezone:
 30    """
 31    Parse a timezone offset string in the format '+HHMM' or '-HHMM' into a timezone object.
 32
 33    >>> parse_timezone_offset('+0530')
 34    timezone(timedelta(seconds=19800))
 35    """
 36    match = re.match(r"([+-])(\d{2})(\d{2})", offset_str)
 37    if not match:
 38        raise ValueError(f"Invalid timezone offset format: {offset_str}")
 39    sign, hours, minutes = match.groups()
 40    hours, minutes = int(hours), int(minutes)
 41    offset = timedelta(hours=hours, minutes=minutes)
 42
 43    if sign == "-":
 44        offset = -offset
 45
 46    return timezone(offset)
 47
 48
 49class _WocObject:
 50    _ident: str
 51    """Identifier of the object"""
 52    woc: WocMapsBase
 53    """WocMap instance"""
 54    key: str
 55    """Key of the object"""
 56
 57    def __init__(
 58        self,
 59        *args,
 60        woc: Optional[WocMapsBase] = None,
 61        **kwargs,
 62    ):
 63        self.woc = woc or _global_woc
 64        assert (
 65            self.woc is not None
 66        ), "WocMaps not initialized: call init_woc_objects() or supply a woc keyword argument"
 67
 68    def __repr__(self) -> str:
 69        return f"{self.__class__.__name__}({self.key})"
 70
 71    def __str__(self) -> str:
 72        return self.key
 73
 74    def __eq__(self, value: object) -> bool:
 75        if not isinstance(value, self.__class__):
 76            return False
 77        return self.key == value.key
 78
 79    @property
 80    def hash(self) -> str:
 81        return hex(hash(self))[2:]
 82
 83    def _get_list_values(self, map_name: str):
 84        """A thin wrapper around WocMapsBase.get_values to handle KeyError"""
 85        try:
 86            return self.woc.get_values(map_name, self.key)
 87        except KeyError:
 88            return []
 89
 90
 91class _GitObject(_WocObject):
 92    """Base class for SHA1-indexed Git objects (commit, tree, blob)"""
 93
 94    def __init__(
 95        self,
 96        key: str,
 97        *args,
 98        woc: Optional[WocMapsBase] = None,
 99        **kwargs,
100    ):
101        super().__init__(*args, woc=woc, **kwargs)
102        assert len(key) == 40, "SHA1 hash must be 40 characters long"
103        self.key = key
104
105    @cached_property
106    def data(self):
107        obj = self.__class__.__name__.lower()
108        return self.woc.show_content(obj, self.key)
109
110    def __hash__(self):
111        return int(self.key, 16)
112
113    @property
114    def hash(self) -> str:
115        return self.key
116
117
118class _NamedObject(_WocObject):
119    """Base class for objects indexed by a string key"""
120
121    def __init__(
122        self,
123        key: str,
124        *args,
125        woc: Optional[WocMapsBase] = None,
126        **kwargs,
127    ):
128        super().__init__(*args, woc=woc, **kwargs)
129        self.key = key
130
131    def __hash__(self):
132        return fnvhash(self.key.encode())
133
134
135class Author(_NamedObject):
136    _ident = "a"
137
138    @cached_property
139    def _username_email(self) -> Tuple[str, str]:
140        _splited = self.key.split(" <", 1)
141        if len(_splited) == 1:
142            return _splited[0], ""
143        return _splited[0], _splited[1][:-1]
144
145    @property
146    def name(self) -> str:
147        return self._username_email[0]
148
149    @property
150    def email(self) -> str:
151        return self._username_email[1]
152
153    @cached_property
154    def blobs(self) -> "List[Blob]":
155        return [Blob(b) for b in self._get_list_values(f"{self._ident}2b")]
156
157    @cached_property
158    def commits(self) -> "List[Commit]":
159        return [Commit(c) for c in self._get_list_values(f"{self._ident}2c")]
160
161    @cached_property
162    def files(self) -> "List[File]":
163        return [File(f) for f in self._get_list_values(f"{self._ident}2f")]
164
165    @cached_property
166    def projects(self) -> "List[Project]":
167        return [Project(p) for p in self._get_list_values(f"{self._ident}2p")]
168
169    @cached_property
170    def unique_authors(self) -> List["UniqueAuthor"]:
171        return [UniqueAuthor(a) for a in self._get_list_values(f"{self._ident}2A")]
172
173    @property
174    def authors(self):
175        raise NotImplementedError("Author object does not have authors method")
176
177    @property
178    def aliases(self) -> List["Author"]:
179        _unique_authors = self.unique_authors
180        if len(_unique_authors) == 0:
181            return []
182        return _unique_authors[0].authors
183
184    @cached_property
185    def first_blobs(self) -> List["Blob"]:
186        return [Blob(b) for b in self._get_list_values(f"{self._ident}2fb")]
187
188
189class UniqueAuthor(Author):
190    _ident = "A"
191
192    @property
193    def unique_authors(self) -> "List[Author]":
194        raise NotImplementedError(
195            "UniqueAuthor object does not have unique_authors method"
196        )
197
198    @cached_property
199    def authors(self) -> "List[Author]":
200        return [Author(a) for a in self._get_list_values(f"{self._ident}2a")]
201
202
203class Blob(_GitObject):
204    _ident = "b"
205
206    @cached_property
207    def _pos(self) -> Tuple[int, int]:
208        return self.woc._get_pos("blob", self.key)
209
210    def __len__(self) -> int:
211        return self._pos[1]
212
213    def __str__(self) -> str:
214        return self.data
215
216    @cached_property
217    def commits(self) -> "List[Commit]":
218        return [Commit(sha) for sha in self._get_list_values("b2c")]
219
220    @cached_property
221    def first_author(self) -> "Tuple[datetime, Author, Commit]":
222        """
223        Returns the timestamp, author, and commit of the first author.
224
225        >>> woc.get_values('b2fa', '05fe634ca4c8386349ac519f899145c75fff4169'))
226        (datetime.datetime(2014, 9, 7, 2, 59, 48), Author(Audris Mockus <audris@utk.edu>), Commit(e4af89166a17785c1d741b8b1d5775f3223f510f))
227        """
228        _out = self.woc.get_values("b2fa", self.key)
229        _date = datetime.fromtimestamp(int(_out[0]))
230        _author = Author(_out[1])
231        _commit = Commit(_out[2])
232        return _date, _author, _commit
233
234    @cached_property
235    def time_author_commits(self) -> "List[Tuple[datetime, Author, Commit]]":
236        _out = self._get_list_values("b2tac")
237        return [
238            (datetime.fromtimestamp(int(d[0])), Author(d[1]), Commit(d[2])) for d in _out
239        ]
240
241    @cached_property
242    def files(self) -> "List[File]":
243        return [File(f) for f in self._get_list_values("b2f")]
244
245    @cached_property
246    def projects_unique(self) -> "List[RootProject]":
247        return [RootProject(p) for p in self._get_list_values("b2P")]
248
249    @cached_property
250    def changed_from(self) -> "List[Tuple[Blob, Commit, File]]":
251        return [
252            (Blob(b), Commit(c), File(f)) for b, c, f in self._get_list_values("bb2cf")
253        ]
254
255    @cached_property
256    def changed_to(self) -> "List[Tuple[Blob, Commit, File]]":
257        return [
258            (Blob(b), Commit(c), File(f)) for b, c, f in self._get_list_values("obb2cf")
259        ]
260
261
262class Commit(_GitObject):
263    _ident = "c"
264
265    @cached_property
266    def data_obj(self):
267        _ret = {}
268        (
269            _ret["tree"],
270            _ret["parent"],
271            (_ret["author"], _ret["author_timestamp"], _ret["author_timezone"]),
272            (_ret["committer"], _ret["committer_timestamp"], _ret["committer_timezone"]),
273            _ret["message"],
274        ) = self.data
275        return _ret
276
277    @property
278    def author(self) -> Author:
279        return Author(self.data_obj["author"])
280
281    @property
282    def authored_at(self) -> datetime:
283        tz = parse_timezone_offset(self.data_obj["author_timezone"])
284        return datetime.fromtimestamp(int(self.data_obj["author_timestamp"]), tz)
285
286    @property
287    def committer(self) -> Author:
288        return Author(self.data_obj["committer"])
289
290    @property
291    def committed_at(self) -> datetime:
292        tz = parse_timezone_offset(self.data_obj["committer_timezone"])
293        return datetime.fromtimestamp(int(self.data_obj["committer_timestamp"]), tz)
294
295    @property
296    def full_message(self) -> str:
297        """Full message of the commit"""
298        return self.data_obj["message"]
299
300    @property
301    def message(self) -> str:
302        """Short message of the commit"""
303        return self.data_obj["message"].split("\n", 1)[0]
304
305    @cached_property
306    def tree(self) -> "Tree":
307        return Tree(self.data_obj["tree"])
308
309    @property
310    def _parent_shas(self) -> List[str]:
311        return self.data_obj["parent"]
312
313    @property
314    def parents(self) -> List["Commit"]:
315        """Parent commits of this commit"""
316        return [Commit(p) for p in self.data_obj["parent"]]
317
318    @cached_property
319    def projects(self) -> List["Project"]:
320        """Projects associated with this commit"""
321        return [Project(p) for p in self._get_list_values("c2p")]
322
323    @cached_property
324    def root_projects(self) -> List["RootProject"]:
325        """Root projects associated with this commit"""
326        return [RootProject(p) for p in self._get_list_values("c2P")]
327
328    @cached_property
329    def children(self) -> List["Commit"]:
330        """Children of this commit"""
331        return [Commit(c) for c in self._get_list_values("c2cc")]
332
333    @cached_property
334    def _file_names(self) -> List[str]:
335        return self._get_list_values("c2f")
336
337    @cached_property
338    def _file_set(self) -> Set[str]:
339        return set(self._file_names)
340
341    @cached_property
342    def files(self) -> List["File"]:
343        """Files changed in this commit"""
344        return [File(f) for f in self._file_names]
345
346    @cached_property
347    def _blob_shas(self) -> List[str]:
348        return self._get_list_values("c2b")
349
350    @cached_property
351    def _blob_set(self) -> Set[str]:
352        return set(self._blob_shas)
353
354    @cached_property
355    def blobs(self) -> List["Blob"]:
356        """
357        Blobs changed in this commit.
358
359        This relation is known to miss every first file in all trees.
360        Consider using Commit.tree.blobs as a slower but more accurate
361        alternative.
362        """
363        return [Blob(b) for b in self._get_list_values("c2b")]
364
365    @cached_property
366    def time_author(self) -> Tuple[datetime, Author]:
367        """Timestamp and author of the commit"""
368        res = self.woc.get_values("c2ta", self.key)
369        return datetime.fromtimestamp(int(res[0])), Author(res[1])
370
371    @cached_property
372    def root(self) -> "Tuple[Commit, int]":
373        """Root commit of the project"""
374        sha, dis = self.woc.get_values("c2r", self.key)
375        return Commit(sha), int(dis)
376
377    @cached_property
378    def changeset(self) -> "List[Tuple[File, Blob, Blob]]":
379        """Returns changed files, their new and old blobs"""
380        return [
381            (File(f), Blob(new), Blob(old))
382            for f, new, old in self._get_list_values("c2fbb")
383        ]
384
385    def compare(
386        self, parent: Union["Commit", str], threshold=0.5
387    ) -> Generator[
388        Tuple[Optional["File"], Optional["File"], Optional["Blob"], Optional["Blob"]],
389        None,
390        None,
391    ]:
392        """
393        Compare two Commits.
394
395        :param parent: another commit to compare to.
396                Expected order is `diff = child_commit - parent_commit`
397
398        :return: a generator of 4-tuples `(old_path, new_path, old_sha, new_sha)`
399
400        Examples:
401        - a new file 'setup.py' was created:
402            `(None, 'setup.py', None, 'file_sha')`
403        - an existing 'setup.py' was deleted:
404            `('setup.py', None, 'old_file_sha', None)`
405        - setup.py.old was renamed to setup.py, content unchanged:
406            `('setup.py.old', 'setup.py', 'file_sha', 'file_sha')`
407        - setup.py was edited:
408            `('setup.py', 'setup.py', 'old_file_sha', 'new_file_sha')`
409        - setup.py.old was edited and renamed to setup.py:
410            `('setup.py.old', 'setup.py', 'old_file_sha', 'new_file_sha')`
411
412        Detecting the last one is computationally expensive. You can adjust this
413        behaviour by passing the `threshold` parameter, which is 0.5 by default.
414        It means that if roughly 50% of the file content is the same,
415        it is considered a match. `threshold=1` means that only exact
416        matches are considered, effectively disabling this comparison.
417        If threshold is set to 0, any pair of deleted and added file will be
418        considered renamed and edited; this last case doesn't make much sense so
419        don't set it too low.
420        """
421        if isinstance(parent, str):
422            parent = Commit(parent)
423        if not isinstance(parent, Commit):
424            raise TypeError("parent must be a Commit or a commit hash")
425
426        # # filename: (blob sha before, blob sha after)
427        # new_files = self.tree._file_blob_map
428        # new_paths = self.tree._file_set
429        # old_files = parent.tree._file_blob_map
430        # old_paths = parent.tree._file_set
431
432        # !!! We really need to traverse the trees ###
433        new_files: Dict[File, Blob] = {}
434        for f, b in self.tree.traverse():
435            new_files[f] = b
436        old_files: Dict[File, Blob] = {}
437        for f, b in parent.tree.traverse():
438            old_files[f] = b
439
440        # unchanged_paths
441        for f in new_files.keys() & old_files.keys():
442            if new_files[f] != old_files[f]:
443                # i.e. Blob sha Changed!
444                yield f, f, old_files[f], new_files[f]
445
446        added_paths: Set[File] = new_files.keys() - old_files.keys()
447        deleted_paths: Set[File] = old_files.keys() - new_files.keys()
448
449        if threshold >= 1:  # i.e. only exact matches are considered
450            for f in added_paths:  # add
451                yield None, f, None, new_files[f]
452            for f in deleted_paths:
453                yield f, None, old_files[f], None
454            return
455
456        if parent.hash not in self._parent_shas:
457            warnings.warn(
458                "Comparing non-adjacent commits might be "
459                "computationally expensive. Proceed with caution."
460            )
461
462        # search for matches
463        sm = difflib.SequenceMatcher()
464        # for each added blob, try to find a match in deleted blobs
465        #   if there is a match, signal a rename and remove from deleted
466        #   if there is no match, signal a new file
467        # unused deleted blobs are indeed deleted
468        for added_file, added_blob in new_files.items():
469            sm.set_seq1(added_blob.data)
470            matched = False
471            for deleted_file, deleted_blob in old_files.items():
472                sm.set_seq2(deleted_blob.data)
473                # use quick checks first (lower bound by length diff)
474                if (
475                    sm.real_quick_ratio() > threshold
476                    and sm.quick_ratio() > threshold
477                    and sm.ratio() > threshold
478                ):
479                    yield deleted_file, added_file, deleted_blob, added_blob
480                    del old_files[deleted_file]
481                    matched = True
482                    break
483            if not matched:  # this is a new file
484                yield None, added_file, None, added_blob
485
486        for deleted_file, deleted_blob in old_files.items():
487            yield deleted_file, None, deleted_blob, None
488
489    def __sub__(self, parent: "Commit"):
490        return self.compare(parent)
491
492
493class File(_NamedObject):
494    _ident = "f"
495
496    @property
497    def path(self) -> str:
498        return self.key
499
500    @property
501    def name(self) -> str:
502        return self.key.split("/")[-1]
503
504    @cached_property
505    def authors(self) -> List[Author]:
506        return [Author(a) for a in self._get_list_values("f2a")]
507
508    @cached_property
509    def blobs(self) -> List[Blob]:
510        return [Blob(b) for b in self._get_list_values("f2b")]
511
512    @cached_property
513    def commits(self) -> List[Commit]:
514        return [Commit(c) for c in self._get_list_values("f2c")]
515
516
517class Tree(_GitObject):
518    _ident = "t"
519
520    @cached_property
521    def data(self) -> str:
522        return self.woc.show_content("tree", self.key)
523
524    @property
525    def _file_names(self) -> List[str]:
526        return [l[1] for l in self.data]
527
528    @cached_property
529    def _file_set(self) -> Set[str]:
530        return {l[1] for l in self.data}
531
532    @property
533    def files(self) -> List["File"]:
534        return [File(f) for f in self._file_names]
535
536    @property
537    def _blob_shas(self) -> List[str]:
538        return [l[2] for l in self.data]
539
540    @cached_property
541    def _blob_set(self) -> Set[str]:
542        return {l[2] for l in self.data}
543
544    @property
545    def blobs(self) -> List["Blob"]:
546        return [Blob(b) for b in self._blob_shas]
547
548    @cached_property
549    def _file_blob_map(self) -> Dict[str, str]:
550        return {l[1]: l[2] for l in self.data}
551
552    def _traverse(self) -> "Generator[Tuple[str, str], None, None]":
553        for mode, fname, sha in self.data:
554            # trees are always 40000:
555            # https://stackoverflow.com/questions/1071241
556            if mode != "40000":
557                yield fname, sha
558            else:
559                _logger.debug(f"traverse: into {fname} ({sha})")
560                for _fname, _sha in Tree(sha)._traverse():
561                    yield fname + "/" + _fname, _sha
562
563    def traverse(self) -> "Generator[Tuple[File, Blob], None, None]":
564        for fname, sha in self._traverse():
565            yield File(fname), Blob(sha)
566
567    def __contains__(self, item: Union[str, File, Blob]) -> bool:
568        if isinstance(item, str):
569            return item in self._file_names or item in self._blob_shas
570        if isinstance(item, File):
571            return item.text in self._file_names
572        if isinstance(item, Blob):
573            return item.hex in self._blob_shas
574        return False
575
576    def __str__(self) -> str:
577        return "\n".join([" ".join(l) for l in self.data])
578
579    def __len__(self) -> int:
580        return len(self.data)
581
582    def __iter__(self) -> "Generator[Tuple[File, Blob], None, None]":
583        for l in self.data:
584            yield File(l[1]), Blob(l[2])
585
586
587class Project(_NamedObject):
588    _ident = "p"
589
590    @cached_property
591    def _platform_repo(self) -> str:
592        URL_PREFIXES = self.woc.config["sites"]
593        prefix, body = self.key.split("_", 1)
594        if prefix == "sourceforge.net":
595            platform = URL_PREFIXES[prefix]
596        elif prefix in URL_PREFIXES and "_" in body:
597            platform = URL_PREFIXES[prefix]
598            body = body.replace("_", "/", 1)
599        elif "." in prefix:
600            platform = prefix
601            body = body.replace("_", "/", 1)
602        else:
603            platform = "github.com"
604            body = self.key.replace("_", "/", 1)
605        return platform, body
606
607    @property
608    def url(self) -> str:
609        """
610        Get the URL for a given project URI.
611
612        >>> Project('CS340-19_lectures').url
613        'http://github.com/CS340-19/lectures'
614        """
615        platform, body = self._platform_repo
616        URL_PREFIXES = self.woc.config["sites"]
617        if platform in URL_PREFIXES:
618            return f"https://{URL_PREFIXES[platform]}/{body}"
619        return f"https://{platform}/{body}"
620
621    @cached_property
622    def authors(self) -> "List[Author]":
623        return [Author(a) for a in self._get_list_values(f"{self._ident}2a")]
624
625    @cached_property
626    def _commit_shas(self) -> "List[str]":
627        return self._get_list_values(f"{self._ident}2c")
628
629    @cached_property
630    def _commit_set(self) -> "Set[str]":
631        return self._commit_map.keys()
632
633    @cached_property
634    def _commit_map(self) -> "Dict[str, Commit]":
635        return {c.hash: c for c in self.commits}
636
637    @cached_property
638    def commits(self) -> "List[Commit]":
639        return [Commit(c) for c in self._commit_shas]
640
641    @cached_property
642    def root_projects(self) -> "List[RootProject]":
643        return [RootProject(p) for p in self._get_list_values(f"{self._ident}2P")]
644
645    def __contains__(self, item: Union[str, Commit]) -> bool:
646        if isinstance(item, str):
647            return item in self._commit_set
648        elif isinstance(item, Commit):
649            return item.hash in self._commit_set
650        return False
651
652    @cached_property
653    def head(self) -> "Commit":
654        """
655        Get the HEAD commit of the repository.
656
657        >>> Project('user2589_minicms').head
658        Commit(f2a7fcdc51450ab03cb364415f14e634fa69b62c)
659        >>> Project('RoseTHERESA_SimpleCMS').head
660        Commit(a47afa002ccfd3e23920f323b172f78c5c970250)
661        """
662        # Sometimes (very rarely) commit dates are wrong, so the latest commit
663        # is not actually the head. The magic below is to account for this
664        parents = set().union(*(c._parent_shas for c in self.commits))
665        heads = [self._commit_map[c] for c in self._commit_set - parents]
666
667        # it is possible that there is more than one head.
668        # E.g. it happens when HEAD is moved manually (git reset)
669        # and continued with a separate chain of commits.
670        # in this case, let's just use the latest one
671        # actually, storing refs would make it much simpler
672        _heads_sorted = sorted(heads, key=lambda c: c.authored_at or _DAY_Z, reverse=True)
673        if len(_heads_sorted) == 0:
674            raise ValueError("No head commit found")
675        return _heads_sorted[0]
676
677    @cached_property
678    def tail(self) -> "Commit":
679        """
680        Get the first commit SHA by following first parents.
681
682        >>> Project(b'user2589_minicms').tail
683        Commit(1e971a073f40d74a1e72e07c682e1cba0bae159b)
684        """
685        pts = {c._parent_shas[0] for c in self.commits if c._parent_shas}
686        for c in self.commits:
687            if c.hash in pts and not c._parent_shas:
688                return c
689
690    @cached_property
691    def earliest_commit(self) -> "Commit":
692        """Get the earliest commit of the repository"""
693        return min(self.commits, key=lambda c: c.authored_at or _DAY_Z)
694
695    @cached_property
696    def latest_commit(self) -> "Commit":
697        """Get the latest commit of the repository"""
698        return max(self.commits, key=lambda c: c.authored_at or _DAY_Z)
699
700    def commits_fp(self) -> Generator["Commit", None, None]:
701        """
702        Get a commit chain by following only the first parent.
703
704        Mimic https://git-scm.com/docs/git-log#git-log---first-parent.
705        Thus, you only get a small subset of the full commit tree.
706
707        >>> p = Project(b'user2589_minicms')
708        >>> set(c.sha for c in p.commits_fp).issubset(p.commit_shas)
709        True
710
711        In scenarios where branches are not important, it can save a lot
712        of computing.
713
714        Yields:
715            Commit: binary commit shas, following first parent only,
716                from the latest to the earliest.
717        """
718        # Simplified version of self.head():
719        #   - slightly less precise,
720        #   - 20% faster
721        #
722        # out of 500 randomly sampled projects, 493 had the same head.
723        # In the remaining 7:
724        #     2 had the same commit chain length,
725        #     3 had one more commit
726        #     1 had two more commits
727        #     1 had three more commits
728        # Execution time:
729        #   simplified version (argmax): ~153 seconds
730        #   self.head(): ~190 seconds
731
732        # at this point we know all commits are in the dataset
733        # (validated in __iter___)
734        commit = self.latest_commit
735
736        while commit:
737            # no point try-except: the truth value of a list is len(list)
738            first_parent = commit._parent_shas and commit._parent_shas[0]
739            yield commit
740            if not first_parent:
741                break
742            commit = self._commit_map.get(first_parent, Commit(first_parent))
743
744    def __iter__(self) -> "Generator[Commit, None, None]":
745        for c in self.commits:
746            try:
747                if c.author in self.woc.config["ignoredAuthors"]:
748                    continue
749                yield c
750            except KeyError:
751                pass
752
753    @property
754    def projects(self) -> List["Project"]:
755        raise NotImplementedError("Project object does not have projects method")
756
757    def download_blob(self, blob_sha: str) -> str:
758        """
759        Download the blob content from remote.
760        """
761        try:
762            from urllib.parse import quote_plus
763
764            import requests
765        except ImportError:
766            raise ImportError(
767                "This function requires the requests module. Install it via `pip install requests`"
768            )
769
770        if self._platform_repo[0] == "github.com":
771            project = self._platform_repo[1]
772            _r = requests.get(
773                f"https://api.github.com/repos/{project}/git/blobs/{blob_sha}",
774                allow_redirects=True,
775                headers={"Accept": "application/vnd.github.raw+json"},
776            )
777            _r.raise_for_status()
778            return _r.content
779        elif self._platform_repo[0] == "gitlab.com":
780            if not hasattr(self, "gitlab_project_id"):
781                project = quote_plus(self._platform_repo[1])
782                r = requests.get(f"https://gitlab.com/api/v4/projects/{project}")
783                r.raise_for_status()
784                self.gitlab_project_id = r.json()["id"]
785            _r = requests.get(
786                f"https://gitlab.com/api/v4/projects/{self.gitlab_project_id}/repository/blobs/{blob_sha}/raw",
787                allow_redirects=True,
788            )
789            _r.raise_for_status()
790            return _r.content
791        else:
792            raise NotImplementedError(
793                "The function is not implemented for " + self._platform_repo[0]
794            )
795
796    def save(self, path: str, commit: Optional[Commit] = None):
797        """
798        Save the project files to the disk. Binary blobs are retrieved from the remote.
799
800        :param path: The path to save the files.
801        :param commit: Save the files at this commit. If None, the head or latest commit is used.
802        """
803        if commit is None:
804            try:
805                commit = self.head
806            except ValueError:
807                _logger.warning(
808                    f"No head commit found for {self.key}, using latest commit"
809                )
810                commit = self.latest_commit
811
812        flist = list(commit.tree.traverse())
813        for idx, (f, blob) in enumerate(flist):
814            _logger.debug(f"{idx + 1}/{len(flist)}: {f.path}")
815            _p = os.path.join(path, f.path)
816            os.makedirs(os.path.dirname(_p), exist_ok=True)
817            with open(_p, "wb") as f:
818                try:
819                    f.write(blob.data.encode())
820                except KeyError:
821                    _logger.info(f"Missing blob {blob.key}")
822                    try:
823                        if self._platform_repo[0] in ("github.com", "gitlab.com"):
824                            f.write(self.download_blob(blob.hash))
825                    except Exception as e:
826                        _logger.error(f"Failed to download blob {blob.hash}: {e}")
827                except Exception as e:
828                    _logger.error(f"Failed to write blob {blob.hash}: {e}")
829
830
831class RootProject(Project):
832    _ident = "P"
833
834    @cached_property
835    def unique_authors(self) -> "List[Author]":
836        return [UniqueAuthor(a) for a in self._get_list_values(f"{self._ident}2A")]
837
838    @cached_property
839    def commits(self) -> "List[Commit]":
840        return [Commit(c) for c in self._get_list_values(f"{self._ident}2C")]
841
842    @cached_property
843    def projects(self) -> "List[Project]":
844        return [Project(p) for p in self._get_list_values(f"{self._ident}2p")]
845
846    @property
847    def root_projects(self) -> List["RootProject"]:
848        raise NotImplementedError("RootProject object does not have root_projects method")