from __future__ import annotations import calendar import os import re from datetime import date, datetime, timedelta, timezone from pathlib import Path from typing import Any APP_DIR = Path(__file__).resolve().parent STATIC_DIR = APP_DIR / "static" DATA_DIR = APP_DIR / "data" ENV_FILE = APP_DIR / ".env" MENTOR_SKILLS_DIR = APP_DIR / "游资skills" PRIVATE_MENTOR_SKILLS_DIR = DATA_DIR / "private-mentor-skills" TOKEN_PATTERN = re.compile(r"^[A-Za-z0-9_-]{20,128}$") USERNAME_PATTERN = re.compile(r"^[A-Za-z0-9_\-\u4e00-\u9fff]{3,30}$") SESSION_COOKIE = "xiaobai_session" SESSION_MAX_AGE = 30 * 24 * 60 * 60 def load_local_env() -> None: if not ENV_FILE.exists(): return for raw_line in ENV_FILE.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) def save_local_env(updates: dict[str, str]) -> None: values: dict[str, str] = {} if ENV_FILE.exists(): for raw_line in ENV_FILE.read_text(encoding="utf-8").splitlines(): if "=" in raw_line and not raw_line.lstrip().startswith("#"): key, value = raw_line.split("=", 1) values[key.strip()] = value.strip().strip('"').strip("'") values.update(updates) ENV_FILE.write_text( "".join(f"{key}={value}\n" for key, value in values.items()), encoding="utf-8", ) def remove_local_env(keys: set[str]) -> None: if not ENV_FILE.exists(): return kept = [] for raw_line in ENV_FILE.read_text(encoding="utf-8").splitlines(): if "=" in raw_line and not raw_line.lstrip().startswith("#"): key = raw_line.split("=", 1)[0].strip() if key in keys: continue kept.append(raw_line) ENV_FILE.write_text("".join(f"{line}\n" for line in kept), encoding="utf-8") for key in keys: os.environ.pop(key, None) def normalize_date(value: str) -> str: compact = value.replace("-", "").strip() try: parsed = datetime.strptime(compact, "%Y%m%d") except ValueError as exc: raise ValueError("日期格式应为 YYYY-MM-DD。") from exc if parsed.date() > date.today(): raise ValueError("不能查询未来日期。") return parsed.strftime("%Y%m%d") def validate_stock_code(value: str) -> str: code = value.strip() if not re.fullmatch(r"\d{6}", code): raise ValueError("股票代码应为 6 位数字。") return code def tushare_code(code: str) -> str: if code.startswith(("4", "8", "9")): suffix = "BJ" elif code.startswith("6"): suffix = "SH" else: suffix = "SZ" return f"{code}.{suffix}" def validate_text(value: Any, label: str, maximum: int, required: bool = False) -> str: text = str(value or "").strip() if required and not text: raise ValueError(f"{label}不能为空。") if len(text) > maximum: raise ValueError(f"{label}不能超过 {maximum} 个字符。") return text def parse_iso_datetime(value: Any) -> datetime | None: text = str(value or "").strip() if not text: return None try: parsed = datetime.fromisoformat(text) except ValueError: return None return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc) def membership_boundary(value: Any, end: bool) -> str | None: text = str(value or "").strip() if not text: return None try: day = datetime.strptime(text, "%Y-%m-%d").replace(tzinfo=timezone.utc) except ValueError as exc: raise ValueError("会员日期格式应为 YYYY-MM-DD。") from exc if end: day += timedelta(days=1) return day.isoformat(timespec="seconds") def add_months(value: datetime, months: int) -> datetime: month_index = value.year * 12 + value.month - 1 + months year, zero_based_month = divmod(month_index, 12) month = zero_based_month + 1 day = min(value.day, calendar.monthrange(year, month)[1]) return value.replace(year=year, month=month, day=day)