migration: preserve startup accounts and system slice
This commit is contained in:
+2
-128
@@ -1,129 +1,3 @@
|
||||
from __future__ import annotations
|
||||
"""Compatibility imports for code that still uses the original configuration module."""
|
||||
|
||||
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)
|
||||
from backend.bootstrap.config import * # noqa: F401,F403
|
||||
|
||||
Reference in New Issue
Block a user