forked from importers/git-importer
74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
import itertools
|
|
import logging
|
|
import re
|
|
|
|
from lib.obs_revision import OBSRevision
|
|
|
|
|
|
class History:
|
|
"""Store the history of revisions of a package in different
|
|
projects.
|
|
|
|
"""
|
|
|
|
def __init__(self, obs, package):
|
|
self.obs = obs
|
|
self.package = package
|
|
|
|
self.revisions = {}
|
|
|
|
def __contains__(self, project):
|
|
return project in self.revisions
|
|
|
|
def __getitem__(self, project):
|
|
return self.revisions[project]
|
|
|
|
def _extract_copypac(self, comment):
|
|
original_project = re.findall(
|
|
r"osc copypac from project:(.*) package:", comment
|
|
)
|
|
return original_project[0] if original_project else None
|
|
|
|
def sort_all_revisions(self):
|
|
"""Sort revisions for all projects, from older to newer"""
|
|
return sorted(itertools.chain(*self.revisions.values()), key=lambda x: x.time)
|
|
|
|
def find_revision(self, project, revisionid, accepted_at):
|
|
last_commited_revision = None
|
|
for r in self.revisions.get(project, []):
|
|
logging.debug(f"Find revision {revisionid} [{accepted_at}]: {r}")
|
|
if str(r.rev) == str(revisionid) or r.srcmd5 == revisionid:
|
|
if r.ignored:
|
|
logging.debug(
|
|
f"{r} fits but is ignored, returning {last_commited_revision}"
|
|
)
|
|
return last_commited_revision
|
|
else:
|
|
logging.debug(f"{r} fits")
|
|
return r
|
|
if r.time > accepted_at:
|
|
# if we can't find the right revision, we take the last
|
|
# commit. Before ~2012 the data was tracked really loosely
|
|
# (e.g. using different timezones and the state field was
|
|
# only introduced in 2016...)
|
|
logging.warning(
|
|
f"Deploying workaround for missing request revision - returning {last_commited_revision}"
|
|
)
|
|
return last_commited_revision
|
|
if r.commit:
|
|
last_commited_revision = r
|
|
logging.info("No commited revision found, returning None")
|
|
return None
|
|
|
|
def find_last_rev_after_time(self, project, time):
|
|
# revs = self.projects.get(project, [])
|
|
# return next((r for r in reversed(revs) if r.time <= time), None)
|
|
prev = None
|
|
for rev in self.revisions.get(project, []):
|
|
if rev.time > time:
|
|
return prev
|
|
if rev.time == time:
|
|
return rev
|
|
prev = rev
|
|
return prev
|