dev

domain-api/app/core/__init__.py

@@ -0,0 +1 @@
+

domain-api/app/core/config.py

@@ -0,0 +1,51 @@
+from pathlib import Path
+
+from pydantic import field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+BASE_DIR = Path(__file__).resolve().parents[2]
+WORKSPACE_DIR = BASE_DIR.parent
+DOMAINCHECK_DIR = WORKSPACE_DIR / "domainCheck"
+API_ENV_FILE = BASE_DIR / ".env"
+ENV_FILE = DOMAINCHECK_DIR / ".env"
+
+
+class Settings(BaseSettings):
+    api_prefix: str = "/api/v1"
+    api_host: str = "0.0.0.0"
+    api_port: int = 8100
+    cors_origins: list[str] = ["http://127.0.0.1:3200", "http://localhost:3200"]
+    db_host: str = "127.0.0.1"
+    db_port: int = 5432
+    db_database: str = "domain"
+    db_user: str = "postgres"
+    db_password: str = "postgres"
+    redis_host: str = "127.0.0.1"
+    redis_port: int = 6379
+    redis_password: str = ""
+    redis_db: int = 0
+    domain_root: str = str(DOMAINCHECK_DIR)
+    admin_username: str = "admin"
+    admin_password: str = "admin"
+    worker_mode: str = "windows-local"
+    worker_service_name: str = "domaincheck-worker"
+    api_service_name: str = "domaincheck-api"
+
+    @field_validator("cors_origins", mode="before")
+    @classmethod
+    def parse_cors_origins(cls, value: object) -> object:
+        if isinstance(value, str):
+            if value.strip().startswith("["):
+                return value
+            return [item.strip() for item in value.split(",") if item.strip()]
+        return value
+
+    model_config = SettingsConfigDict(
+        env_file=(str(API_ENV_FILE), str(ENV_FILE)),
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+
+settings = Settings()

domain-api/app/core/db.py

@@ -0,0 +1,20 @@
+from contextlib import contextmanager
+
+import psycopg2
+
+from app.core.config import settings
+
+
+@contextmanager
+def get_db():
+    conn = psycopg2.connect(
+        host=settings.db_host,
+        port=settings.db_port,
+        dbname=settings.db_database,
+        user=settings.db_user,
+        password=settings.db_password,
+    )
+    try:
+        yield conn
+    finally:
+        conn.close()

domain-api/app/core/files.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from datetime import datetime
+
+from app.core.config import settings
+
+
+def domain_root() -> Path:
+    return Path(settings.domain_root)
+
+
+def read_json(relative_path: str, default: dict | list | None = None):
+    path = domain_root() / relative_path
+    if not path.exists():
+        return {} if default is None else default
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def write_json(relative_path: str, payload) -> None:
+    path = domain_root() / relative_path
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as handle:
+        json.dump(payload, handle, ensure_ascii=False, indent=2)
+
+
+def tail_lines(relative_path: str, max_lines: int = 120) -> list[str]:
+    path = domain_root() / relative_path
+    if not path.exists():
+        return []
+    with path.open("r", encoding="utf-8", errors="replace") as handle:
+        return handle.read().splitlines()[-max_lines:]
+
+
+def runtime_root() -> Path:
+    path = Path(__file__).resolve().parents[2] / "runtime"
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def read_runtime_json(filename: str, default: dict | list | None = None):
+    path = runtime_root() / filename
+    if not path.exists():
+        return {} if default is None else default
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def write_runtime_json(filename: str, payload) -> None:
+    path = runtime_root() / filename
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as handle:
+        json.dump(payload, handle, ensure_ascii=False, indent=2, default=str)
+
+
+def exports_root() -> Path:
+    path = runtime_root() / "exports"
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def import_root() -> Path:
+    path = runtime_root() / "imports"
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def settings_backup_root() -> Path:
+    path = runtime_root() / "settings_backups"
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def load_export_records() -> list[dict]:
+    path = runtime_root() / "export_tasks.json"
+    if not path.exists():
+        return []
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def save_export_record(record: dict) -> None:
+    records = load_export_records()
+    records.insert(0, record)
+    path = runtime_root() / "export_tasks.json"
+    with path.open("w", encoding="utf-8") as handle:
+        json.dump(records[:200], handle, ensure_ascii=False, indent=2, default=str)
+
+
+def load_import_records() -> list[dict]:
+    path = runtime_root() / "import_tasks.json"
+    if not path.exists():
+        return []
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def save_import_records(records: list[dict]) -> None:
+    path = runtime_root() / "import_tasks.json"
+    with path.open("w", encoding="utf-8") as handle:
+        json.dump(records[:200], handle, ensure_ascii=False, indent=2, default=str)
+
+
+def timestamp_filename(prefix: str, ext: str) -> str:
+    return f"{prefix}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.{ext}"

domain-api/app/core/redis_client.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+import redis
+
+from app.core.config import settings
+
+
+def get_redis() -> redis.Redis:
+    return redis.Redis(
+        host=settings.redis_host,
+        port=settings.redis_port,
+        password=settings.redis_password or None,
+        db=settings.redis_db,
+        decode_responses=True,
+        socket_connect_timeout=5,
+        socket_timeout=5,
+    )