majuna/app/models/base.py

97 lines
3.3 KiB
Python

from datetime import datetime
from typing import List
from app.extensions import db
from app.models import AbstractConfiguration
class Group(AbstractConfiguration):
group_name = db.Column(db.String(80), unique=True, nullable=False)
eotk = db.Column(db.Boolean())
origins = db.relationship("Origin", back_populates="group")
bridgeconfs = db.relationship("BridgeConf", back_populates="group")
eotks = db.relationship("Eotk", back_populates="group")
onions = db.relationship("Onion", back_populates="group")
smart_proxies = db.relationship("SmartProxy", back_populates="group")
pools = db.relationship("Pool", secondary="pool_group", back_populates="groups")
@classmethod
def csv_header(cls) -> List[str]:
return super().csv_header() + [
"group_name", "eotk"
]
class Pool(AbstractConfiguration):
pool_name = db.Column(db.String(80), unique=True, nullable=False)
@classmethod
def csv_header(cls) -> List[str]:
return super().csv_header() + [
"pool_name"
]
proxies = db.relationship("Proxy", back_populates="pool")
lists = db.relationship("MirrorList", back_populates="pool")
groups = db.relationship("Group", secondary="pool_group", back_populates="pools")
class PoolGroup(db.Model): # type: ignore[name-defined,misc]
pool_id = db.Column(db.Integer, db.ForeignKey("pool.id"), primary_key=True)
group_id = db.Column(db.Integer, db.ForeignKey("group.id"), primary_key=True)
class MirrorList(AbstractConfiguration):
pool_id = db.Column(db.Integer, db.ForeignKey("pool.id"))
provider = db.Column(db.String(255), nullable=False)
format = db.Column(db.String(20), nullable=False)
encoding = db.Column(db.String(20), nullable=False)
container = db.Column(db.String(255), nullable=False)
branch = db.Column(db.String(255), nullable=False)
role = db.Column(db.String(255), nullable=True)
filename = db.Column(db.String(255), nullable=False)
pool = db.relationship("Pool", back_populates="lists")
providers_supported = {
"github": "GitHub",
"gitlab": "GitLab",
"http_post": "HTTP POST",
"s3": "AWS S3",
}
formats_supported = {
"bc2": "Bypass Censorship v2",
"bc3": "Bypass Censorship v3",
"bca": "Bypass Censorship Analytics",
"bridgelines": "Tor Bridge Lines"
}
encodings_supported = {
"json": "JSON (Plain)",
"jsno": "JSON (Obfuscated)",
"js": "JavaScript (Plain)",
"jso": "JavaScript (Obfuscated)"
}
def destroy(self) -> None:
self.destroyed = datetime.utcnow()
self.updated = datetime.utcnow()
def url(self) -> str:
if self.provider == "gitlab":
return f"https://gitlab.com/{self.container}/-/raw/{self.branch}/{self.filename}"
if self.provider == "github":
return f"https://raw.githubusercontent.com/{self.container}/{self.branch}/{self.filename}"
if self.provider == "s3":
return f"s3://{self.container}/{self.filename}"
if self.provider == "http_post":
return str(self.container)
return "Unknown provider"
@classmethod
def csv_header(cls) -> List[str]:
return super().csv_header() + [
"provider", "format", "container", "branch", "filename"
]