bellingcat/auto-archiver-api
1
0
Fork 0
mirror of https://github.com/bellingcat/auto-archiver-api.git synced 2026-06-14 22:48:35 +03:00
Code Issues Packages Projects Releases Wiki Activity

Format and lint shared directory (#64)

Browse Source
This commit is contained in:
Michael Plunkett
2025-03-03 13:20:50 -06:00
committed by GitHub
parent a9ca410d08
commit 1ca0ae2fb2
13 changed files with 255 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
app/shared/user_groups.py
View File

@@ -19,13 +19,16 @@ class UserGroups:
user_groups = self.read_yaml(filename)
self.validate_and_load(user_groups)
def read_yaml(self, user_groups_filename):
@staticmethod
def read_yaml(user_groups_filename):
# read yaml safely
with open(user_groups_filename) as inf:
try:
return yaml.safe_load(inf)
except yaml.YAMLError as e:
logger.error(f"could not open user groups filename {user_groups_filename}: {e}")
logger.error(
f"could not open user groups filename {user_groups_filename}: {e}"
)
raise e
def validate_and_load(self, user_groups):
@@ -52,22 +55,36 @@ class GroupPermissions(BaseModel):
max_monthly_mbs: int = 0
priority: str = "low"
@field_validator('max_sheets', 'max_archive_lifespan_months', 'max_monthly_urls', 'max_monthly_mbs', mode='before')
@classmethod
@field_validator(
"max_sheets",
"max_archive_lifespan_months",
"max_monthly_urls",
"max_monthly_mbs",
mode="before",
)
def validate_max_values(cls, v):
if v < -1:
raise ValueError("max_* values should be positive integers or -1 (for no limit).")
raise ValueError(
"max_* values should be positive integers or -1 (for no limit)."
)
return v
@field_validator('sheet_frequency', mode='before')
@classmethod
@field_validator("sheet_frequency", mode="before")
def validate_sheet_frequency(cls, v):
if not v: return []
if not v:
return []
allowed = ["daily", "hourly"]
for k in v:
if k not in allowed:
raise ValueError(f"Invalid sheet frequency: '{k}', expected one of {allowed}")
raise ValueError(
f"Invalid sheet frequency: '{k}', expected one of {allowed}"
)
return v
@field_validator('priority', mode='before')
@classmethod
@field_validator("priority", mode="before")
def validate_priority(cls, v):
v = v.lower()
if v not in ["low", "high"]:
@@ -81,7 +98,8 @@ class GroupModel(BaseModel):
orchestrator_sheet: str
permissions: GroupPermissions
@field_validator('orchestrator', 'orchestrator_sheet', mode='before')
@classmethod
@field_validator("orchestrator", "orchestrator_sheet", mode="before")
def validate_orchestrator(cls, v):
if not os.path.exists(v):
raise ValueError(f"Orchestrator file not found with this path: {v}")
@@ -105,13 +123,17 @@ class GroupModel(BaseModel):
service_account_json = find_service_account_email(orch)
if not service_account_json:
raise ValueError(f"service_account key not found in orchestrator sheet file: {self.orchestrator_sheet}.")
raise ValueError(
f"service_account key not found in orchestrator sheet file: {self.orchestrator_sheet}."
)
with open(service_account_json) as f:
self._service_account_email = json.load(f).get("client_email")
if not self._service_account_email:
raise ValueError(f"Service account email not found in {service_account_json}.")
raise ValueError(
f"Service account email not found in {service_account_json}."
)
return self._service_account_email
@@ -121,29 +143,45 @@ class UserGroupModel(BaseModel):
domains: Dict[str, List[str]] = Field(default_factory=dict)
groups: Dict[str, GroupModel] = Field(default_factory=dict)
@field_validator('users', mode='before')
@classmethod
@field_validator("users", mode="before")
def validate_emails(cls, v):
for email in v.keys():
if '@' not in email:
raise ValueError(f"Invalid user, it should be an address: {email}")
if "@" not in email:
raise ValueError(
f"Invalid user, it should be an address: {email}"
)
if not v[email]:
raise ValueError(f"User {email} has no explicitly listed groups, only include them here if they should be in a group.")
raise ValueError(
f"User {email} has no explicitly listed groups, only include them here if they should be in a group."
)
# all users belong to the default group
return {k.lower().strip(): list(set(["default"] + [g.lower().strip() for g in v])) for k, v in v.items()}
return {
k.lower().strip(): list(
set(["default"] + [g.lower().strip() for g in v])
)
for k, v in v.items()
}
@field_validator('domains', mode='before')
@classmethod
@field_validator("domains", mode="before")
def validate_domains(cls, v):
for domain, members in v.items():
if '.' not in domain:
raise ValueError(f"Invalid domain, it should contain a dot: {domain}")
if "." not in domain:
raise ValueError(
f"Invalid domain, it should contain a dot: {domain}"
)
if not members:
raise ValueError(f"Domain {domain} should have at least one member.")
return {k.lower().strip(): list(set([g.lower().strip() for g in v])) for k, v in v.items()}
raise ValueError(
f"Domain {domain} should have at least one member."
)
return {
k.lower().strip(): list({[g.lower().strip() for g in v]})
for k, v in v.items()
}
@field_validator('groups', mode='before')
@classmethod
@field_validator("groups", mode="before")
def validate_groups(cls, v):
if "default" not in v.keys():
raise ValueError("Please include a 'default' group.")
@@ -154,20 +192,28 @@ class UserGroupModel(BaseModel):
raise ValueError(f"Group names should be lowercase: {group}")
return v
@model_validator(mode='after')
@model_validator(mode="after")
def check_groups_consistency(self) -> Self:
groups_in_domains = set([g for domain in self.domains for g in self.domains[domain]])
groups_in_users = set([g for user in self.users for g in self.users[user]])
groups_in_domains = {
g for domain in self.domains for g in self.domains[domain]
}
groups_in_users = {g for user in self.users for g in self.users[user]}
configured_groups = set(self.groups.keys())
# groups mentioned in domains and users should be defined, but this is not a ValueError since historical DB data may require it
# groups mentioned in domains and users should be defined, but this is
# not a ValueError since historical DB data may require it
if groups_in_domains - configured_groups:
logger.warning(f"These groups are associated to DOMAINS but not defined in the GROUPS section, the domains settings may not work as expected: {groups_in_domains - configured_groups}")
logger.warning(
f"These groups are associated to DOMAINS but not defined in the GROUPS section, the domains settings may not work as expected: {groups_in_domains - configured_groups}"
)
if groups_in_users - configured_groups:
logger.warning(f"These groups are associated to USERS but not defined in the GROUPS section, the users settings may not work as expected: {groups_in_users - configured_groups}")
logger.warning(
f"These groups are associated to USERS but not defined in the GROUPS section, the users settings may not work as expected: {groups_in_users - configured_groups}"
)
return self
# for the API return values