jasima/majuna
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: remove pydantic from list generation

Browse source
This commit is contained in:
Iain Learmonth 2024-11-09 11:08:48 +00:00
parent 1e70ec8fa6
commit d08388c339
8 changed files with 164 additions and 197 deletions
Download patch file Download diff file Expand all files Collapse all files

77
app/lists/bc2.py
View file

@ -1,38 +1,28 @@
# pylint: disable=too-few-public-methods
import builtins
from datetime import datetime
from typing import List, Dict, Union, Any, Optional
from pydantic import BaseModel, Field
from typing import List, Optional, TypedDict
from app.models.base import Pool
from app.models.mirrors import Origin, Proxy
class BC2Alternative(BaseModel):
class BC2Alternative(TypedDict):
proto: str
type: str
created_at: datetime
updated_at: datetime
created_at: str
updated_at: str
url: str
class BC2Site(BaseModel):
main_domain: str = Field(description="The main domain name of the website, excluding \"www.\" if present.",
examples=["bbc.co.uk", "bbc.com", "guardianproject.info"])
class BC2Site(TypedDict):
main_domain: str
available_alternatives: List[BC2Alternative]
class BypassCensorship2(BaseModel):
version: str = Field(description="Version number of the Bypass Censorship Extension schema in use", )
class BypassCensorship2(TypedDict):
version: str
sites: List[BC2Site]
class Config:
title = "Bypass Censorship Version 2"
def onion_alternative(origin: Origin) -> List[Dict[str, Any]]:
def onion_alternative(origin: Origin) -> List[BC2Alternative]:
url: Optional[str] = origin.onion()
if url is None:
return []
@ -41,22 +31,23 @@ def onion_alternative(origin: Origin) -> List[Dict[str, Any]]:
"type": "eotk",
"created_at": str(origin.added),
"updated_at": str(origin.updated),
"url": url}
]
"url": url
}]
def proxy_alternative(proxy: Proxy) -> Dict[str, Any]:
def proxy_alternative(proxy: Proxy) -> Optional[BC2Alternative]:
if proxy.url is None:
return None
return {
"proto": "https",
"type": "mirror",
"created_at": str(proxy.added),
"updated_at": str(proxy.updated),
"created_at": proxy.added.isoformat(),
"updated_at": proxy.updated.isoformat(),
"url": proxy.url
}
def main_domain(origin: Origin) -> str:
# Both description and domain_name are required to be not null in the database schema
description: str = origin.description
if description.startswith("proxy:"):
return description[len("proxy:"):].replace("www.", "")
@ -65,20 +56,30 @@ def main_domain(origin: Origin) -> str:
def active_proxies(origin: Origin, pool: Pool) -> List[Proxy]:
def _filter_fn(proxy: Proxy) -> bool:
return proxy.url is not None and not proxy.deprecated and not proxy.destroyed and proxy.pool_id == pool.id
return list(filter(_filter_fn, origin.proxies))
return [
proxy for proxy in origin.proxies
if proxy.url is not None and not proxy.deprecated and not proxy.destroyed and proxy.pool_id == pool.id
]
def mirror_sites(pool: Pool) -> Dict[
str, Union[str, List[Dict[str, Union[str, List[Dict[str, str]]]]]]]:
return {"version": "2.0", "sites": [{"main_domain": main_domain(origin),
"available_alternatives": onion_alternative(origin) + [
proxy_alternative(proxy) for proxy in
active_proxies(origin, pool)]} for origin in
Origin.query.order_by(Origin.domain_name).all() if
origin.destroyed is None]}
def mirror_sites(pool: Pool) -> BypassCensorship2:
origins = Origin.query.filter(Origin.destroyed.is_(None)).order_by(Origin.domain_name).all()
sites: List[BC2Site] = []
for origin in origins:
# Gather alternatives, filtering out None values from proxy_alternative
alternatives = onion_alternative(origin) + [
alt for proxy in active_proxies(origin, pool)
if (alt := proxy_alternative(proxy)) is not None
]
if getattr(builtins, "__sphinx_build__", False):
schema = BypassCensorship2.schema_json()
# Add the site dictionary to the list
sites.append({
"main_domain": main_domain(origin),
"available_alternatives": list(alternatives)
})
return {
"version": "2.0",
"sites": sites
}