feat(share): T7.1 短链接生成 (base62 + SQLite 映射表)
- data/share/short_link.py (657 行)：ShortLinkService 核心模块 - base62 编解码 + secrets.choice 加密随机短码 - SQLite WAL 模式持久化，share_links 表 + 3 索引 - 访问控制：permission (read/comment/edit/admin) + sha256 密码 + expires_at + revoked - 完整 CRUD: create/get/resolve/revoke/list_by_*/get_stats/purge_expired - route_short_link() 路由辅助供 Flask/FastAPI 等挂载 /s/<code> - data/share/tests/test_short_link.py (447 行)：25 个 pytest 用例覆盖全部 API - scripts/gaokao-shortlink (271 行)：CLI 入口 (create/resolve/revoke/list/stats/purge)
This commit is contained in:
hermes-coder
1
data/share/__init__.py
Normal file
1
data/share/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""分享数据模块 (T7)"""
|
||||
657
data/share/short_link.py
Normal file
657
data/share/short_link.py
Normal file
@@ -0,0 +1,657 @@
|
||||
"""
|
||||
高考志愿填报系统 - 短链接生成服务 (T7.1)
|
||||
|
||||
提供:
|
||||
- 短码生成 (base62, 默认 6 位)
|
||||
- SQLite 映射表 (短码 → 报告元数据)
|
||||
- 访问控制 (有效期 / 密码 / 权限)
|
||||
- 访问统计 (次数 / 时间)
|
||||
|
||||
URL 模式:
|
||||
/s/{code} → 分享短链接 (由 Web 路由 /s/<code> 调用 resolve())
|
||||
/s/ABC123 → 短码示例
|
||||
|
||||
依赖:
|
||||
仅 Python 3.8+ 标准库 (sqlite3, hashlib, secrets, base64, binascii)
|
||||
"""
|
||||
|
||||
import binascii
|
||||
import hashlib
|
||||
import os
|
||||
import secrets
|
||||
import sqlite3
|
||||
import string
|
||||
import time
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 常量
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# base62 字母表 (0-9, A-Z, a-z) — URL-safe, 无需 urlencode
|
||||
BASE62_ALPHABET = string.digits + string.ascii_uppercase + string.ascii_lowercase
|
||||
BASE62_LEN = len(BASE62_ALPHABET) # 62
|
||||
|
||||
# 短码默认长度 (6 位 = 56B 空间, 实际使用远小于该值, 碰撞概率极低)
|
||||
DEFAULT_CODE_LEN = 6
|
||||
|
||||
# 默认数据库路径
|
||||
DEFAULT_DB_PATH = Path(__file__).resolve().parent / "short_links.db"
|
||||
|
||||
# 权限级别
|
||||
PERM_READ = "read"
|
||||
PERM_COMMENT = "comment"
|
||||
PERM_EDIT = "edit"
|
||||
PERM_ADMIN = "admin"
|
||||
VALID_PERMISSIONS = {PERM_READ, PERM_COMMENT, PERM_EDIT, PERM_ADMIN}
|
||||
|
||||
# 状态常量
|
||||
STATUS_OK = "ok"
|
||||
STATUS_NOT_FOUND = "not_found"
|
||||
STATUS_REVOKED = "revoked"
|
||||
STATUS_EXPIRED = "expired"
|
||||
STATUS_PASSWORD_REQUIRED = "password_required"
|
||||
STATUS_PASSWORD_WRONG = "password_wrong"
|
||||
STATUS_PASSWORD_WRONG = "password_wrong"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 数据类
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class ShareLink:
|
||||
"""分享链接记录"""
|
||||
|
||||
code: str
|
||||
report_id: str
|
||||
owner_id: str = "anonymous"
|
||||
permission: str = PERM_COMMENT
|
||||
password_hash: Optional[str] = None # sha256 hex
|
||||
expires_at: Optional[float] = None # unix timestamp
|
||||
revoked: int = 0
|
||||
access_count: int = 0
|
||||
last_access_at: Optional[float] = None
|
||||
created_at: float = field(default_factory=time.time)
|
||||
note: Optional[str] = None
|
||||
|
||||
def is_expired(self, now: Optional[float] = None) -> bool:
|
||||
if self.expires_at is None:
|
||||
return False
|
||||
return (now or time.time()) >= self.expires_at
|
||||
|
||||
def is_active(self) -> bool:
|
||||
return self.revoked == 0 and not self.is_expired()
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = asdict(self)
|
||||
d["created_at_iso"] = _iso(self.created_at)
|
||||
if self.expires_at is not None:
|
||||
d["expires_at_iso"] = _iso(self.expires_at)
|
||||
else:
|
||||
d["expires_at_iso"] = None
|
||||
if self.last_access_at is not None:
|
||||
d["last_access_at_iso"] = _iso(self.last_access_at)
|
||||
else:
|
||||
d["last_access_at_iso"] = None
|
||||
return d
|
||||
|
||||
|
||||
@dataclass
|
||||
class ResolveResult:
|
||||
"""resolve() 的返回结果"""
|
||||
|
||||
status: str
|
||||
code: str
|
||||
link: Optional[ShareLink] = None
|
||||
reason: str = ""
|
||||
|
||||
@property
|
||||
def ok(self) -> bool:
|
||||
return self.status == STATUS_OK
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = {"status": self.status, "code": self.code, "reason": self.reason}
|
||||
if self.link is not None:
|
||||
d["link"] = self.link.to_dict()
|
||||
return d
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# base62 编解码
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def base62_encode(num: int) -> str:
|
||||
"""
|
||||
把非负整数编码为 base62 字符串
|
||||
例: 0 -> "0", 61 -> "Z", 62 -> "10", 1234567 -> "3Gtv"
|
||||
"""
|
||||
if num < 0:
|
||||
raise ValueError("num must be >= 0")
|
||||
if num == 0:
|
||||
return BASE62_ALPHABET[0]
|
||||
parts = []
|
||||
while num > 0:
|
||||
num, rem = divmod(num, BASE62_LEN)
|
||||
parts.append(BASE62_ALPHABET[rem])
|
||||
return "".join(reversed(parts))
|
||||
|
||||
|
||||
def base62_decode(s: str) -> int:
|
||||
"""把 base62 字符串解码为整数"""
|
||||
if not s:
|
||||
raise ValueError("empty string")
|
||||
n = 0
|
||||
for ch in s:
|
||||
idx = BASE62_ALPHABET.find(ch)
|
||||
if idx < 0:
|
||||
raise ValueError(f"invalid base62 char: {ch!r}")
|
||||
n = n * BASE62_LEN + idx
|
||||
return n
|
||||
|
||||
|
||||
def generate_code(length: int = DEFAULT_CODE_LEN) -> str:
|
||||
"""
|
||||
用加密安全的随机数生成短码
|
||||
注: 大批量时实际碰撞概率可用生日悖论估算
|
||||
(N=10M, length=6, p≈4e-7), 失败后由调用方重试
|
||||
"""
|
||||
if length < 4 or length > 16:
|
||||
raise ValueError("length must be in [4, 16]")
|
||||
return "".join(secrets.choice(BASE62_ALPHABET) for _ in range(length))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 内部工具
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _iso(ts: float) -> str:
|
||||
"""unix timestamp -> ISO8601 (UTC)"""
|
||||
return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()
|
||||
|
||||
|
||||
def _now() -> float:
|
||||
return time.time()
|
||||
|
||||
|
||||
def _hash_password(password: str) -> str:
|
||||
"""密码哈希: sha256 (无盐, 因密码空间足够大; 真实部署可换 argon2)"""
|
||||
if not password:
|
||||
raise ValueError("password must be non-empty")
|
||||
return hashlib.sha256(password.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def _row_to_link(row: sqlite3.Row) -> ShareLink:
|
||||
return ShareLink(
|
||||
code=row["code"],
|
||||
report_id=row["report_id"],
|
||||
owner_id=row["owner_id"],
|
||||
permission=row["permission"],
|
||||
password_hash=row["password_hash"],
|
||||
expires_at=row["expires_at"],
|
||||
revoked=row["revoked"],
|
||||
access_count=row["access_count"],
|
||||
last_access_at=row["last_access_at"],
|
||||
created_at=row["created_at"],
|
||||
note=row["note"],
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 短链接服务
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ShortLinkService:
|
||||
"""
|
||||
短链接服务, 对应 /s/{code} 路由
|
||||
|
||||
用法:
|
||||
svc = ShortLinkService() # 默认 DB 路径
|
||||
link = svc.create(report_id="R-2026-001",
|
||||
permission="read",
|
||||
ttl_days=7)
|
||||
print(link.code) # "aB3xY7"
|
||||
print(svc.url(link.code, base="https://gk.example.com"))
|
||||
# -> "https://gk.example.com/s/aB3xY7"
|
||||
|
||||
result = svc.resolve(link.code)
|
||||
assert result.ok
|
||||
assert result.link.report_id == "R-2026-001"
|
||||
"""
|
||||
|
||||
SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS share_links (
|
||||
code TEXT PRIMARY KEY,
|
||||
report_id TEXT NOT NULL,
|
||||
owner_id TEXT NOT NULL DEFAULT 'anonymous',
|
||||
permission TEXT NOT NULL DEFAULT 'comment',
|
||||
password_hash TEXT,
|
||||
expires_at REAL,
|
||||
revoked INTEGER NOT NULL DEFAULT 0,
|
||||
access_count INTEGER NOT NULL DEFAULT 0,
|
||||
last_access_at REAL,
|
||||
created_at REAL NOT NULL,
|
||||
note TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_share_links_report
|
||||
ON share_links(report_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_share_links_owner
|
||||
ON share_links(owner_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_share_links_expires
|
||||
ON share_links(expires_at);
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: Optional[os.PathLike] = None):
|
||||
self.db_path = Path(db_path) if db_path else DEFAULT_DB_PATH
|
||||
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
# 第一次实例化即建表 (亦可显式调用 init_schema)
|
||||
self.init_schema()
|
||||
|
||||
# ---- 生命周期 ----
|
||||
|
||||
def init_schema(self) -> None:
|
||||
with self._connect() as conn:
|
||||
conn.executescript(self.SCHEMA)
|
||||
conn.commit()
|
||||
|
||||
def _connect(self) -> sqlite3.Connection:
|
||||
# in-memory DB 需要 shared cache 才能在多次 connect 中共享
|
||||
if str(self.db_path) == ":memory:":
|
||||
conn = sqlite3.connect(
|
||||
"file::memory:?cache=shared",
|
||||
timeout=10.0,
|
||||
isolation_level=None,
|
||||
uri=True,
|
||||
)
|
||||
else:
|
||||
conn = sqlite3.connect(
|
||||
str(self.db_path), timeout=10.0, isolation_level=None
|
||||
)
|
||||
conn.row_factory = sqlite3.Row
|
||||
if str(self.db_path) != ":memory:":
|
||||
conn.execute("PRAGMA journal_mode = WAL;")
|
||||
conn.execute("PRAGMA foreign_keys = ON;")
|
||||
return conn
|
||||
|
||||
# ---- 创建 ----
|
||||
|
||||
def create(
|
||||
self,
|
||||
report_id: str,
|
||||
owner_id: str = "anonymous",
|
||||
permission: str = PERM_COMMENT,
|
||||
password: Optional[str] = None,
|
||||
ttl_seconds: Optional[int] = None,
|
||||
ttl_days: Optional[int] = None,
|
||||
code_length: int = DEFAULT_CODE_LEN,
|
||||
max_retries: int = 8,
|
||||
note: Optional[str] = None,
|
||||
) -> ShareLink:
|
||||
"""
|
||||
创建一条分享链接
|
||||
|
||||
参数:
|
||||
report_id 关联的报告 ID (T1 / T2 / T3 报告)
|
||||
owner_id 创建者 ID
|
||||
permission read / comment / edit / admin
|
||||
password 可选访问密码
|
||||
ttl_seconds / ttl_days 二选一, None 表示永不过期
|
||||
code_length 短码长度 (4-16)
|
||||
max_retries 碰撞重试次数
|
||||
note 备注
|
||||
"""
|
||||
if not report_id:
|
||||
raise ValueError("report_id is required")
|
||||
if permission not in VALID_PERMISSIONS:
|
||||
raise ValueError(f"permission must be one of {sorted(VALID_PERMISSIONS)}")
|
||||
if ttl_seconds is not None and ttl_days is not None:
|
||||
raise ValueError("ttl_seconds 与 ttl_days 不能同时指定")
|
||||
if ttl_seconds is not None and ttl_seconds <= 0:
|
||||
raise ValueError("ttl_seconds must be > 0")
|
||||
if ttl_days is not None and ttl_days <= 0:
|
||||
raise ValueError("ttl_days must be > 0")
|
||||
|
||||
expires_at = None
|
||||
if ttl_seconds is not None:
|
||||
expires_at = _now() + ttl_seconds
|
||||
elif ttl_days is not None:
|
||||
expires_at = _now() + ttl_days * 86400
|
||||
|
||||
password_hash = _hash_password(password) if password else None
|
||||
|
||||
last_err: Optional[Exception] = None
|
||||
for _ in range(max_retries):
|
||||
code = generate_code(code_length)
|
||||
try:
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO share_links(
|
||||
code, report_id, owner_id, permission,
|
||||
password_hash, expires_at, revoked,
|
||||
access_count, last_access_at, created_at, note
|
||||
) VALUES (?, ?, ?, ?, ?, ?, 0, 0, NULL, ?, ?)
|
||||
""",
|
||||
(
|
||||
code,
|
||||
report_id,
|
||||
owner_id,
|
||||
permission,
|
||||
password_hash,
|
||||
expires_at,
|
||||
_now(),
|
||||
note,
|
||||
),
|
||||
)
|
||||
return self.get(code)
|
||||
except sqlite3.IntegrityError as e:
|
||||
# 唯一冲突: 碰撞, 重试
|
||||
last_err = e
|
||||
continue
|
||||
raise RuntimeError(
|
||||
f"无法生成唯一短码 (length={code_length}, retries={max_retries}): {last_err}"
|
||||
)
|
||||
|
||||
# ---- 查询 ----
|
||||
|
||||
def get(self, code: str) -> Optional[ShareLink]:
|
||||
if not code:
|
||||
return None
|
||||
with self._connect() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT * FROM share_links WHERE code = ?", (code,)
|
||||
).fetchone()
|
||||
return _row_to_link(row) if row else None
|
||||
|
||||
def resolve(
|
||||
self,
|
||||
code: str,
|
||||
password: Optional[str] = None,
|
||||
record_access: bool = True,
|
||||
) -> ResolveResult:
|
||||
"""
|
||||
解析短码 -> (status, link, reason)
|
||||
|
||||
状态:
|
||||
ok / not_found / revoked / expired /
|
||||
password_required / password_wrong
|
||||
|
||||
校验顺序 (按 HTTP 语义):
|
||||
1. 存在?
|
||||
2. 已撤销?
|
||||
3. 已过期?
|
||||
4. 需要密码?
|
||||
5. 密码正确?
|
||||
"""
|
||||
link = self.get(code)
|
||||
if link is None:
|
||||
return ResolveResult(
|
||||
status=STATUS_NOT_FOUND, code=code, reason="code not found"
|
||||
)
|
||||
if link.revoked != 0:
|
||||
return ResolveResult(
|
||||
status=STATUS_REVOKED, code=code, link=link, reason="revoked"
|
||||
)
|
||||
if link.is_expired():
|
||||
return ResolveResult(
|
||||
status=STATUS_EXPIRED, code=code, link=link, reason="expired"
|
||||
)
|
||||
if link.password_hash is not None:
|
||||
if not password:
|
||||
return ResolveResult(
|
||||
status=STATUS_PASSWORD_REQUIRED,
|
||||
code=code,
|
||||
link=link,
|
||||
reason="password required",
|
||||
)
|
||||
if _hash_password(password) != link.password_hash:
|
||||
return ResolveResult(
|
||||
status=STATUS_PASSWORD_WRONG,
|
||||
code=code,
|
||||
link=link,
|
||||
reason="wrong password",
|
||||
)
|
||||
|
||||
if record_access:
|
||||
self._bump_access(link.code)
|
||||
link.access_count += 1
|
||||
link.last_access_at = _now()
|
||||
return ResolveResult(status=STATUS_OK, code=code, link=link, reason="ok")
|
||||
|
||||
def _bump_access(self, code: str) -> None:
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE share_links
|
||||
SET access_count = access_count + 1,
|
||||
last_access_at = ?
|
||||
WHERE code = ?
|
||||
""",
|
||||
(_now(), code),
|
||||
)
|
||||
|
||||
# ---- 撤销 ----
|
||||
|
||||
def revoke(self, code: str, owner_id: Optional[str] = None) -> bool:
|
||||
"""
|
||||
撤销一条链接
|
||||
如果指定 owner_id, 只有 owner 匹配时才能撤销 (防止越权)
|
||||
返回: 是否从「未撤销」变为「已撤销」(True/False),
|
||||
重复撤销 / 不存在的 code 均返回 False
|
||||
"""
|
||||
with self._connect() as conn:
|
||||
if owner_id is not None:
|
||||
cur = conn.execute(
|
||||
"UPDATE share_links SET revoked = 1 "
|
||||
"WHERE code = ? AND owner_id = ? AND revoked = 0",
|
||||
(code, owner_id),
|
||||
)
|
||||
else:
|
||||
cur = conn.execute(
|
||||
"UPDATE share_links SET revoked = 1 WHERE code = ? AND revoked = 0",
|
||||
(code,),
|
||||
)
|
||||
return cur.rowcount > 0
|
||||
|
||||
# ---- 列表 / 统计 ----
|
||||
|
||||
def list_by_report(self, report_id: str) -> List[ShareLink]:
|
||||
with self._connect() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM share_links WHERE report_id = ? ORDER BY created_at DESC",
|
||||
(report_id,),
|
||||
).fetchall()
|
||||
return [_row_to_link(r) for r in rows]
|
||||
|
||||
def list_by_owner(self, owner_id: str, limit: int = 100) -> List[ShareLink]:
|
||||
with self._connect() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM share_links WHERE owner_id = ? "
|
||||
"ORDER BY created_at DESC LIMIT ?",
|
||||
(owner_id, limit),
|
||||
).fetchall()
|
||||
return [_row_to_link(r) for r in rows]
|
||||
|
||||
def get_stats(self, code: str) -> Optional[dict]:
|
||||
link = self.get(code)
|
||||
if link is None:
|
||||
return None
|
||||
return {
|
||||
"code": link.code,
|
||||
"report_id": link.report_id,
|
||||
"access_count": link.access_count,
|
||||
"last_access_at": link.last_access_at,
|
||||
"last_access_at_iso": _iso(link.last_access_at)
|
||||
if link.last_access_at
|
||||
else None,
|
||||
"revoked": bool(link.revoked),
|
||||
"expired": link.is_expired(),
|
||||
"created_at_iso": _iso(link.created_at),
|
||||
"expires_at_iso": _iso(link.expires_at) if link.expires_at else None,
|
||||
}
|
||||
|
||||
# ---- 维护 ----
|
||||
|
||||
def purge_expired(self) -> int:
|
||||
"""清理过期记录 (T7.1 不强制, 仅供维护)"""
|
||||
with self._connect() as conn:
|
||||
cur = conn.execute(
|
||||
"DELETE FROM share_links WHERE expires_at IS NOT NULL AND expires_at < ?",
|
||||
(_now(),),
|
||||
)
|
||||
return cur.rowcount
|
||||
|
||||
def count(self, owner_id: Optional[str] = None) -> int:
|
||||
with self._connect() as conn:
|
||||
if owner_id is None:
|
||||
row = conn.execute("SELECT COUNT(*) AS n FROM share_links").fetchone()
|
||||
else:
|
||||
row = conn.execute(
|
||||
"SELECT COUNT(*) AS n FROM share_links WHERE owner_id = ?",
|
||||
(owner_id,),
|
||||
).fetchone()
|
||||
return int(row["n"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 路由辅助 (供 Web 框架 / T7.5 分享页接入)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _route_with_svc(
|
||||
svc: ShortLinkService,
|
||||
code: str,
|
||||
password: Optional[str] = None,
|
||||
base_url: str = "http://localhost:8000",
|
||||
) -> dict:
|
||||
"""route_short_link 的核心逻辑, 接受外部 svc (用于 in-memory 测试)"""
|
||||
res = svc.resolve(code, password=password, record_access=True)
|
||||
out: dict = {
|
||||
"code": code,
|
||||
"status": res.status,
|
||||
"reason": res.reason,
|
||||
"url": f"{base_url.rstrip('/')}/s/{code}",
|
||||
}
|
||||
if res.ok and res.link is not None:
|
||||
out["report_id"] = res.link.report_id
|
||||
out["permission"] = res.link.permission
|
||||
out["owner_id"] = res.link.owner_id
|
||||
out["access_count"] = res.link.access_count
|
||||
return out
|
||||
|
||||
|
||||
def route_short_link(
|
||||
code: str,
|
||||
password: Optional[str] = None,
|
||||
base_url: str = "http://localhost:8000",
|
||||
db_path: Optional[os.PathLike] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
模拟 /s/{code} 路由的入口
|
||||
返回 JSON-friendly dict, 包含 redirect URL 或错误信息
|
||||
|
||||
用法 (Flask 示意):
|
||||
@app.route("/s/<code>")
|
||||
def short_link(code):
|
||||
return jsonify(route_short_link(code, request.args.get("pwd")))
|
||||
|
||||
参数:
|
||||
code 短码 (URL 路径变量)
|
||||
password 访问密码 (query/body)
|
||||
base_url 用于构造 /s/{code} 完整 URL
|
||||
db_path 数据库路径, 默认 DEFAULT_DB_PATH; 测试可覆盖
|
||||
"""
|
||||
svc = ShortLinkService(db_path=db_path)
|
||||
return _route_with_svc(svc, code, password=password, base_url=base_url)
|
||||
|
||||
|
||||
def build_url(code: str, base: str = "http://localhost:8000") -> str:
|
||||
"""生成 /s/{code} 完整 URL"""
|
||||
return f"{base.rstrip('/')}/s/{code}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI 直接调用 (python -m data.share.short_link ...) 用作冒烟
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _self_test() -> None: # pragma: no cover - 仅 CLI 触发
|
||||
svc = ShortLinkService(db_path=":memory:")
|
||||
print("=== short_link self test ===")
|
||||
|
||||
# 1. create
|
||||
link = svc.create(
|
||||
report_id="R-2026-001",
|
||||
owner_id="alice",
|
||||
permission="read",
|
||||
ttl_days=7,
|
||||
)
|
||||
print(f"created: code={link.code} report_id={link.report_id}")
|
||||
assert len(link.code) == DEFAULT_CODE_LEN
|
||||
assert all(c in BASE62_ALPHABET for c in link.code)
|
||||
|
||||
# 2. resolve (ok)
|
||||
res = svc.resolve(link.code)
|
||||
assert res.ok, f"unexpected status: {res.status}"
|
||||
print(f"resolve ok: access_count={res.link.access_count}")
|
||||
|
||||
# 3. revoke
|
||||
assert svc.revoke(link.code, owner_id="alice") is True
|
||||
res = svc.resolve(link.code)
|
||||
assert res.status == STATUS_REVOKED
|
||||
print("revoke works")
|
||||
|
||||
# 4. password
|
||||
link2 = svc.create(
|
||||
report_id="R-2026-002",
|
||||
owner_id="bob",
|
||||
permission="edit",
|
||||
password="s3cr3t",
|
||||
)
|
||||
res = svc.resolve(link2.code, password=None)
|
||||
assert res.status == STATUS_PASSWORD_REQUIRED, res.status
|
||||
res = svc.resolve(link2.code, password="wrong")
|
||||
assert res.status == STATUS_PASSWORD_WRONG
|
||||
res = svc.resolve(link2.code, password="s3cr3t")
|
||||
assert res.ok
|
||||
print("password works")
|
||||
|
||||
# 5. expire
|
||||
link3 = svc.create(
|
||||
report_id="R-2026-003",
|
||||
owner_id="carol",
|
||||
permission="read",
|
||||
ttl_seconds=1,
|
||||
)
|
||||
time.sleep(1.2)
|
||||
res = svc.resolve(link3.code)
|
||||
assert res.status == STATUS_EXPIRED
|
||||
print("expire works")
|
||||
|
||||
# 6. base62 codec sanity
|
||||
for n in [0, 1, 61, 62, 62**6 - 1, 62**6]:
|
||||
s = base62_encode(n)
|
||||
assert base62_decode(s) == n, (n, s)
|
||||
print("base62 codec works")
|
||||
|
||||
# 7. route helper (复用同一 in-memory svc, 避免创建新连接)
|
||||
out = _route_with_svc(
|
||||
svc, link2.code, password="s3cr3t", base_url="https://gk.example.com"
|
||||
)
|
||||
assert out["status"] == "ok", out
|
||||
assert out["url"] == "https://gk.example.com/s/" + link2.code
|
||||
print(f"route helper works: {out['url']}")
|
||||
|
||||
print("=== all self tests passed ===")
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
_self_test()
|
||||
0
data/share/tests/__init__.py
Normal file
0
data/share/tests/__init__.py
Normal file
447
data/share/tests/test_short_link.py
Normal file
447
data/share/tests/test_short_link.py
Normal file
@@ -0,0 +1,447 @@
|
||||
"""
|
||||
短链接服务单元测试 (T7.1)
|
||||
|
||||
运行:
|
||||
python3 -m pytest data/share/tests/test_short_link.py -v
|
||||
# 或 (无 pytest 时)
|
||||
python3 data/share/tests/test_short_link.py
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
# 让 data.share 可被 import
|
||||
PROJ = Path(__file__).resolve().parents[3]
|
||||
sys.path.insert(0, str(PROJ))
|
||||
|
||||
import tempfile
|
||||
import os
|
||||
import uuid
|
||||
|
||||
from data.share.short_link import ( # noqa: E402
|
||||
BASE62_ALPHABET,
|
||||
DEFAULT_CODE_LEN,
|
||||
PERM_COMMENT,
|
||||
PERM_EDIT,
|
||||
PERM_READ,
|
||||
STATUS_EXPIRED,
|
||||
STATUS_NOT_FOUND,
|
||||
STATUS_OK,
|
||||
STATUS_PASSWORD_REQUIRED,
|
||||
STATUS_PASSWORD_WRONG,
|
||||
STATUS_REVOKED,
|
||||
ShortLinkService,
|
||||
base62_decode,
|
||||
base62_encode,
|
||||
build_url,
|
||||
generate_code,
|
||||
route_short_link,
|
||||
)
|
||||
|
||||
# in-memory DB 共享 cache 会跨实例泄漏, 测试中用临时文件更安全
|
||||
_TMP_DBS: list = []
|
||||
|
||||
|
||||
def make_svc() -> ShortLinkService:
|
||||
"""为每个测试创建独立的临时 SQLite 文件 (避免 in-memory 共享)"""
|
||||
fd, db = tempfile.mkstemp(
|
||||
prefix=f"shortlink_test_{uuid.uuid4().hex[:8]}_", suffix=".db"
|
||||
)
|
||||
os.close(fd)
|
||||
_TMP_DBS.append(db)
|
||||
return ShortLinkService(db_path=db)
|
||||
|
||||
|
||||
def cleanup_tmp_dbs():
|
||||
for db in _TMP_DBS:
|
||||
try:
|
||||
os.remove(db)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 简易测试运行器 (兼容无 pytest 环境)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_PASS = 0
|
||||
_FAIL = 0
|
||||
_ERRORS: list = []
|
||||
|
||||
|
||||
def _eq(a, b, msg=""):
|
||||
global _PASS, _FAIL
|
||||
if a == b:
|
||||
_PASS += 1
|
||||
else:
|
||||
_FAIL += 1
|
||||
_ERRORS.append(f"FAIL: {msg or 'equality'}: {a!r} != {b!r}")
|
||||
|
||||
|
||||
def _truthy(v, msg):
|
||||
global _PASS, _FAIL
|
||||
if v:
|
||||
_PASS += 1
|
||||
else:
|
||||
_FAIL += 1
|
||||
_ERRORS.append(f"FAIL: {msg}: {v!r}")
|
||||
|
||||
|
||||
def _raises(fn, exc_type, msg):
|
||||
global _PASS, _FAIL
|
||||
try:
|
||||
fn()
|
||||
except exc_type:
|
||||
_PASS += 1
|
||||
return
|
||||
except Exception as e:
|
||||
_FAIL += 1
|
||||
_ERRORS.append(f"FAIL: {msg}: expected {exc_type}, got {type(e).__name__}: {e}")
|
||||
return
|
||||
_FAIL += 1
|
||||
_ERRORS.append(f"FAIL: {msg}: no exception raised")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# base62 codec
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_base62_codec_basic():
|
||||
_eq(base62_encode(0), "0", "0 -> '0'")
|
||||
# 字母表 = digits + ascii_uppercase + ascii_lowercase
|
||||
# index 0-9 = '0'-'9', 10-35 = 'A'-'Z', 36-61 = 'a'-'z'
|
||||
_eq(base62_encode(9), "9", "9 -> '9'")
|
||||
_eq(base62_encode(10), "A", "10 -> 'A'")
|
||||
_eq(base62_encode(35), "Z", "35 -> 'Z'")
|
||||
_eq(base62_encode(36), "a", "36 -> 'a'")
|
||||
_eq(base62_encode(61), "z", "61 -> 'z'")
|
||||
_eq(base62_encode(62), "10", "62 -> '10'")
|
||||
_eq(base62_encode(62 * 62), "100", "62^2 -> '100'")
|
||||
_truthy(len(base62_encode(62**6 - 1)) == 6, "62^6-1 fits in 6 chars")
|
||||
|
||||
|
||||
def test_base62_codec_roundtrip():
|
||||
for n in [
|
||||
0,
|
||||
1,
|
||||
9,
|
||||
35,
|
||||
36,
|
||||
61,
|
||||
62,
|
||||
62**2,
|
||||
62**3,
|
||||
62**4,
|
||||
62**5,
|
||||
62**6,
|
||||
62**6 + 12345,
|
||||
]:
|
||||
s = base62_encode(n)
|
||||
_eq(base62_decode(s), n, f"roundtrip {n}")
|
||||
|
||||
|
||||
def test_base62_invalid_char():
|
||||
_raises(lambda: base62_decode("abc!"), ValueError, "invalid char raises")
|
||||
_raises(lambda: base62_decode(""), ValueError, "empty raises")
|
||||
_raises(lambda: base62_encode(-1), ValueError, "negative raises")
|
||||
|
||||
|
||||
def test_generate_code():
|
||||
for length in [4, 6, 8, 16]:
|
||||
c = generate_code(length)
|
||||
_eq(len(c), length, f"code length {length}")
|
||||
_truthy(
|
||||
all(ch in BASE62_ALPHABET for ch in c),
|
||||
f"code chars in alphabet (len={length})",
|
||||
)
|
||||
_raises(lambda: generate_code(3), ValueError, "len=3 rejected")
|
||||
_raises(lambda: generate_code(17), ValueError, "len=17 rejected")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 服务: 创建 / 解析
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_create_basic():
|
||||
svc = make_svc()
|
||||
link = svc.create(report_id="R-1", owner_id="alice")
|
||||
_eq(len(link.code), DEFAULT_CODE_LEN, "default code length")
|
||||
_eq(link.report_id, "R-1", "report_id")
|
||||
_eq(link.owner_id, "alice", "owner_id")
|
||||
_eq(link.permission, PERM_COMMENT, "default permission")
|
||||
_eq(link.revoked, 0, "not revoked")
|
||||
_eq(link.access_count, 0, "no access yet")
|
||||
|
||||
|
||||
def test_create_collision_retry():
|
||||
"""验证 create() 在碰撞时会重试 (此处不模拟碰撞, 只验证正常路径)"""
|
||||
svc = make_svc()
|
||||
codes = set()
|
||||
for i in range(50):
|
||||
link = svc.create(report_id=f"R-{i}")
|
||||
_truthy(link.code not in codes, f"code {link.code} unique")
|
||||
codes.add(link.code)
|
||||
|
||||
|
||||
def test_get_missing():
|
||||
svc = make_svc()
|
||||
_eq(svc.get(""), None, "empty code returns None")
|
||||
_eq(svc.get("NOTHERE"), None, "missing code returns None")
|
||||
|
||||
|
||||
def test_resolve_ok():
|
||||
svc = make_svc()
|
||||
link = svc.create(report_id="R-1", owner_id="alice", permission=PERM_READ)
|
||||
res = svc.resolve(link.code)
|
||||
_eq(res.status, STATUS_OK, "resolve ok")
|
||||
_eq(res.link.report_id, "R-1", "resolve report_id")
|
||||
_eq(res.link.access_count, 1, "resolve bumps access_count to 1")
|
||||
_eq(res.link.last_access_at is not None, True, "last_access_at set")
|
||||
|
||||
|
||||
def test_resolve_dry_run():
|
||||
svc = make_svc()
|
||||
link = svc.create(report_id="R-1")
|
||||
res = svc.resolve(link.code, record_access=False)
|
||||
_eq(res.status, STATUS_OK, "dry-run ok")
|
||||
_eq(res.link.access_count, 0, "dry-run keeps access_count 0")
|
||||
|
||||
|
||||
def test_resolve_not_found():
|
||||
svc = make_svc()
|
||||
res = svc.resolve("ZZZZZZ")
|
||||
_eq(res.status, STATUS_NOT_FOUND, "not_found")
|
||||
_eq(res.link, None, "no link object")
|
||||
|
||||
|
||||
def test_resolve_password_required():
|
||||
svc = make_svc()
|
||||
link = svc.create(report_id="R-1", password="s3cr3t")
|
||||
res = svc.resolve(link.code)
|
||||
_eq(res.status, STATUS_PASSWORD_REQUIRED, "password_required")
|
||||
res = svc.resolve(link.code, password="wrong")
|
||||
_eq(res.status, STATUS_PASSWORD_WRONG, "password_wrong")
|
||||
res = svc.resolve(link.code, password="s3cr3t")
|
||||
_eq(res.status, STATUS_OK, "correct pwd ok")
|
||||
|
||||
|
||||
def test_resolve_expired():
|
||||
svc = make_svc()
|
||||
link = svc.create(report_id="R-1", ttl_seconds=1)
|
||||
res = svc.resolve(link.code)
|
||||
_eq(res.status, STATUS_OK, "fresh ok")
|
||||
time.sleep(1.2)
|
||||
res = svc.resolve(link.code)
|
||||
_eq(res.status, STATUS_EXPIRED, "expired after ttl")
|
||||
|
||||
|
||||
def test_resolve_revoked():
|
||||
svc = make_svc()
|
||||
link = svc.create(report_id="R-1", owner_id="alice")
|
||||
_truthy(svc.revoke(link.code, owner_id="alice"), "revoke returns True")
|
||||
res = svc.resolve(link.code)
|
||||
_eq(res.status, STATUS_REVOKED, "revoked")
|
||||
# 重复 revoke
|
||||
_truthy(not svc.revoke(link.code, owner_id="alice"), "double revoke returns False")
|
||||
|
||||
|
||||
def test_revoke_owner_check():
|
||||
svc = make_svc()
|
||||
link = svc.create(report_id="R-1", owner_id="alice")
|
||||
_truthy(not svc.revoke(link.code, owner_id="bob"), "wrong owner rejected")
|
||||
res = svc.resolve(link.code)
|
||||
_eq(res.status, STATUS_OK, "wrong-owner revoke doesn't actually revoke")
|
||||
|
||||
|
||||
def test_ttl_days():
|
||||
svc = make_svc()
|
||||
link = svc.create(report_id="R-1", ttl_days=1)
|
||||
_truthy(link.expires_at is not None, "ttl_days sets expires_at")
|
||||
_truthy(link.expires_at > link.created_at, "expires_at > created_at")
|
||||
|
||||
|
||||
def test_ttl_exclusive():
|
||||
svc = make_svc()
|
||||
_raises(
|
||||
lambda: svc.create(report_id="R-1", ttl_seconds=60, ttl_days=1),
|
||||
ValueError,
|
||||
"ttl_seconds & ttl_days exclusive",
|
||||
)
|
||||
_raises(
|
||||
lambda: svc.create(report_id="R-1", ttl_seconds=0),
|
||||
ValueError,
|
||||
"ttl_seconds > 0",
|
||||
)
|
||||
|
||||
|
||||
def test_invalid_permission():
|
||||
svc = make_svc()
|
||||
_raises(
|
||||
lambda: svc.create(report_id="R-1", permission="superuser"),
|
||||
ValueError,
|
||||
"invalid permission rejected",
|
||||
)
|
||||
|
||||
|
||||
def test_required_report_id():
|
||||
svc = make_svc()
|
||||
_raises(
|
||||
lambda: svc.create(report_id=""),
|
||||
ValueError,
|
||||
"empty report_id rejected",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 列表 / 统计 / 维护
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_list_by_report():
|
||||
svc = make_svc()
|
||||
svc.create(report_id="R-1")
|
||||
svc.create(report_id="R-1", permission=PERM_EDIT)
|
||||
svc.create(report_id="R-2")
|
||||
links = svc.list_by_report("R-1")
|
||||
_eq(len(links), 2, "list_by_report filters correctly")
|
||||
_truthy(
|
||||
all(l.report_id == "R-1" for l in links),
|
||||
"all links belong to R-1",
|
||||
)
|
||||
|
||||
|
||||
def test_list_by_owner():
|
||||
svc = make_svc()
|
||||
svc.create(report_id="R-1", owner_id="alice")
|
||||
svc.create(report_id="R-2", owner_id="alice")
|
||||
svc.create(report_id="R-3", owner_id="bob")
|
||||
links = svc.list_by_owner("alice")
|
||||
_eq(len(links), 2, "alice has 2 links")
|
||||
links = svc.list_by_owner("bob")
|
||||
_eq(len(links), 1, "bob has 1 link")
|
||||
|
||||
|
||||
def test_stats():
|
||||
svc = make_svc()
|
||||
link = svc.create(report_id="R-1")
|
||||
svc.resolve(link.code)
|
||||
svc.resolve(link.code)
|
||||
stats = svc.get_stats(link.code)
|
||||
_truthy(stats is not None, "stats exists")
|
||||
_eq(stats["access_count"], 2, "access_count=2")
|
||||
_eq(stats["code"], link.code, "stats code matches")
|
||||
_eq(svc.get_stats("NOTHERE"), None, "missing stats -> None")
|
||||
|
||||
|
||||
def test_purge_expired():
|
||||
svc = make_svc()
|
||||
svc.create(report_id="R-1", ttl_seconds=1)
|
||||
svc.create(report_id="R-2") # permanent
|
||||
time.sleep(1.2)
|
||||
n = svc.purge_expired()
|
||||
_eq(n, 1, "purged 1 expired")
|
||||
_eq(svc.count(), 1, "permanent link remains")
|
||||
|
||||
|
||||
def test_count():
|
||||
svc = make_svc()
|
||||
_eq(svc.count(), 0, "empty db count=0")
|
||||
svc.create(report_id="R-1", owner_id="alice")
|
||||
svc.create(report_id="R-2", owner_id="alice")
|
||||
svc.create(report_id="R-3", owner_id="bob")
|
||||
_eq(svc.count(), 3, "total 3")
|
||||
_eq(svc.count(owner_id="alice"), 2, "alice 2")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 路由辅助
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_build_url():
|
||||
_eq(
|
||||
build_url("ABC123"),
|
||||
"http://localhost:8000/s/ABC123",
|
||||
"build_url default base",
|
||||
)
|
||||
_eq(
|
||||
build_url("ABC123", base="https://gk.example.com/"),
|
||||
"https://gk.example.com/s/ABC123",
|
||||
"build_url custom base (trailing slash stripped)",
|
||||
)
|
||||
|
||||
|
||||
def test_route_short_link():
|
||||
"""route_short_link 走指定 db; 用 temp file 隔离"""
|
||||
import tempfile
|
||||
import os
|
||||
|
||||
fd, db = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
_TMP_DBS.append(db)
|
||||
try:
|
||||
svc = ShortLinkService(db_path=db)
|
||||
link = svc.create(report_id="R-1", password="s3cr3t")
|
||||
# OK
|
||||
out = route_short_link(
|
||||
link.code,
|
||||
password="s3cr3t",
|
||||
base_url="https://gk.example.com",
|
||||
db_path=db,
|
||||
)
|
||||
_eq(out["status"], STATUS_OK, "route ok")
|
||||
_eq(out["report_id"], "R-1", "route returns report_id")
|
||||
_eq(out["url"], "https://gk.example.com/s/" + link.code, "route url")
|
||||
# password_required
|
||||
out = route_short_link(
|
||||
link.code,
|
||||
base_url="https://gk.example.com",
|
||||
db_path=db,
|
||||
)
|
||||
_eq(out["status"], STATUS_PASSWORD_REQUIRED, "route pwd_required")
|
||||
# not_found
|
||||
out = route_short_link(
|
||||
"ZZZZZZ",
|
||||
base_url="https://gk.example.com",
|
||||
db_path=db,
|
||||
)
|
||||
_eq(out["status"], STATUS_NOT_FOUND, "route not_found")
|
||||
finally:
|
||||
os.remove(db)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def main():
|
||||
test_funcs = [
|
||||
v for k, v in globals().items() if k.startswith("test_") and callable(v)
|
||||
]
|
||||
for fn in test_funcs:
|
||||
try:
|
||||
fn()
|
||||
except Exception as e:
|
||||
global _FAIL
|
||||
_FAIL += 1
|
||||
_ERRORS.append(f"ERROR in {fn.__name__}: {type(e).__name__}: {e}")
|
||||
|
||||
print()
|
||||
print(f"PASS: {_PASS}")
|
||||
print(f"FAIL: {_FAIL}")
|
||||
if _ERRORS:
|
||||
print()
|
||||
for e in _ERRORS:
|
||||
print(f" {e}")
|
||||
sys.exit(1)
|
||||
print("ALL TESTS PASSED")
|
||||
cleanup_tmp_dbs()
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
271
scripts/gaokao-shortlink
Executable file
271
scripts/gaokao-shortlink
Executable file
@@ -0,0 +1,271 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
gaokao-shortlink — T7.1 短链接生成器命令行工具
|
||||
|
||||
子命令:
|
||||
create 创建短链接
|
||||
resolve 解析短链接 (查询/校验)
|
||||
revoke 撤销短链接
|
||||
list 列出某报告 / 某用户的所有链接
|
||||
stats 查看某链接的访问统计
|
||||
purge 清理过期记录 (维护)
|
||||
|
||||
示例:
|
||||
# 创建: 关联报告 R-2026-001, 30 天有效, 只读权限
|
||||
python scripts/gaokao-shortlink create \\
|
||||
--report-id R-2026-001 --owner alice --permission read --ttl-days 30
|
||||
|
||||
# 创建: 带密码
|
||||
python scripts/gaokao-shortlink create \\
|
||||
--report-id R-2026-001 --owner alice --permission comment \\
|
||||
--password s3cr3t --ttl-days 7
|
||||
|
||||
# 解析
|
||||
python scripts/gaokao-shortlink resolve ABC123
|
||||
python scripts/gaokao-shortlink resolve ABC123 --password s3cr3t
|
||||
|
||||
# 撤销
|
||||
python scripts/gaokao-shortlink revoke ABC123 --owner alice
|
||||
|
||||
# 列表
|
||||
python scripts/gaokao-shortlink list --report R-2026-001
|
||||
python scripts/gaokao-shortlink list --owner alice
|
||||
|
||||
# 统计
|
||||
python scripts/gaokao-shortlink stats ABC123
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# 把项目根目录加入 sys.path
|
||||
PROJ_ROOT = Path(__file__).resolve().parent.parent
|
||||
if str(PROJ_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(PROJ_ROOT))
|
||||
|
||||
from data.share.short_link import ( # noqa: E402
|
||||
DEFAULT_DB_PATH,
|
||||
PERM_ADMIN,
|
||||
PERM_COMMENT,
|
||||
PERM_EDIT,
|
||||
PERM_READ,
|
||||
ShortLinkService,
|
||||
STATUS_OK,
|
||||
STATUS_NOT_FOUND,
|
||||
STATUS_PASSWORD_REQUIRED,
|
||||
STATUS_PASSWORD_WRONG,
|
||||
STATUS_REVOKED,
|
||||
STATUS_EXPIRED,
|
||||
VALID_PERMISSIONS,
|
||||
build_url,
|
||||
)
|
||||
|
||||
|
||||
def _out(data, as_json: bool = True) -> None:
|
||||
"""统一输出"""
|
||||
if as_json:
|
||||
print(json.dumps(data, ensure_ascii=False, indent=2, default=str))
|
||||
else:
|
||||
# 人类可读
|
||||
if isinstance(data, dict):
|
||||
for k, v in data.items():
|
||||
print(f"{k}: {v}")
|
||||
else:
|
||||
print(data)
|
||||
|
||||
|
||||
def cmd_create(args) -> int:
|
||||
"""创建短链接"""
|
||||
svc = ShortLinkService(db_path=args.db)
|
||||
try:
|
||||
link = svc.create(
|
||||
report_id=args.report_id,
|
||||
owner_id=args.owner,
|
||||
permission=args.permission,
|
||||
password=args.password,
|
||||
ttl_seconds=args.ttl_seconds,
|
||||
ttl_days=args.ttl_days,
|
||||
code_length=args.length,
|
||||
note=args.note,
|
||||
)
|
||||
except (ValueError, RuntimeError) as e:
|
||||
print(f"创建失败: {e}", file=sys.stderr)
|
||||
return 2
|
||||
out = {
|
||||
"code": link.code,
|
||||
"url": build_url(link.code, base=args.base_url),
|
||||
"report_id": link.report_id,
|
||||
"owner_id": link.owner_id,
|
||||
"permission": link.permission,
|
||||
"expires_at_iso": (
|
||||
link.to_dict()["expires_at_iso"] if link.expires_at else None
|
||||
),
|
||||
"created_at_iso": link.to_dict()["created_at_iso"],
|
||||
"has_password": link.password_hash is not None,
|
||||
}
|
||||
_out(out, as_json=not args.human)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_resolve(args) -> int:
|
||||
"""解析短链接"""
|
||||
svc = ShortLinkService(db_path=args.db)
|
||||
res = svc.resolve(args.code, password=args.password, record_access=not args.dry_run)
|
||||
payload = {
|
||||
"code": res.code,
|
||||
"status": res.status,
|
||||
"reason": res.reason,
|
||||
"url": build_url(res.code, base=args.base_url),
|
||||
}
|
||||
if res.link is not None:
|
||||
payload["report_id"] = res.link.report_id
|
||||
payload["owner_id"] = res.link.owner_id
|
||||
payload["permission"] = res.link.permission
|
||||
payload["access_count"] = res.link.access_count
|
||||
payload["last_access_at_iso"] = (
|
||||
res.link.to_dict()["last_access_at_iso"]
|
||||
if res.link.last_access_at else None
|
||||
)
|
||||
payload["expires_at_iso"] = (
|
||||
res.link.to_dict()["expires_at_iso"]
|
||||
if res.link.expires_at else None
|
||||
)
|
||||
_out(payload, as_json=not args.human)
|
||||
return 0 if res.status == STATUS_OK else 1
|
||||
|
||||
|
||||
def cmd_revoke(args) -> int:
|
||||
"""撤销短链接"""
|
||||
svc = ShortLinkService(db_path=args.db)
|
||||
ok = svc.revoke(args.code, owner_id=args.owner)
|
||||
if not ok:
|
||||
print(
|
||||
f"撤销失败: code={args.code} 不存在或 owner 不匹配",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 1
|
||||
_out({"code": args.code, "revoked": True, "owner": args.owner},
|
||||
as_json=not args.human)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_list(args) -> int:
|
||||
"""列出某报告 / 某用户的链接"""
|
||||
if not args.report and not args.owner:
|
||||
print("必须指定 --report 或 --owner", file=sys.stderr)
|
||||
return 2
|
||||
svc = ShortLinkService(db_path=args.db)
|
||||
if args.report:
|
||||
links = svc.list_by_report(args.report)
|
||||
else:
|
||||
links = svc.list_by_owner(args.owner, limit=args.limit)
|
||||
rows = []
|
||||
for link in links:
|
||||
d = link.to_dict()
|
||||
d["active"] = link.is_active()
|
||||
rows.append(d)
|
||||
_out({"count": len(rows), "links": rows}, as_json=not args.human)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_stats(args) -> int:
|
||||
"""查看某链接访问统计"""
|
||||
svc = ShortLinkService(db_path=args.db)
|
||||
stats = svc.get_stats(args.code)
|
||||
if stats is None:
|
||||
print(f"未找到 code={args.code}", file=sys.stderr)
|
||||
return 1
|
||||
_out(stats, as_json=not args.human)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_purge(args) -> int:
|
||||
"""清理过期记录"""
|
||||
svc = ShortLinkService(db_path=args.db)
|
||||
n = svc.purge_expired()
|
||||
_out({"purged": n}, as_json=not args.human)
|
||||
return 0
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
p = argparse.ArgumentParser(
|
||||
prog="gaokao-shortlink",
|
||||
description="T7.1 短链接生成器 (base62 + SQLite)",
|
||||
)
|
||||
p.add_argument(
|
||||
"--db",
|
||||
default=str(DEFAULT_DB_PATH),
|
||||
help=f"SQLite 数据库路径 (默认: {DEFAULT_DB_PATH})",
|
||||
)
|
||||
p.add_argument(
|
||||
"--base-url",
|
||||
default="http://localhost:8000",
|
||||
help="用于构造 /s/{code} 完整 URL (默认 http://localhost:8000)",
|
||||
)
|
||||
p.add_argument(
|
||||
"--human", action="store_true",
|
||||
help="人类可读输出 (默认 JSON)",
|
||||
)
|
||||
sub = p.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
# ---- create ----
|
||||
pc = sub.add_parser("create", help="创建一条短链接")
|
||||
pc.add_argument("--report-id", required=True, help="关联报告 ID")
|
||||
pc.add_argument("--owner", default="anonymous", help="创建者 ID")
|
||||
pc.add_argument(
|
||||
"--permission", default=PERM_COMMENT, choices=sorted(VALID_PERMISSIONS),
|
||||
help=f"权限 (默认 {PERM_COMMENT}; {', '.join(sorted(VALID_PERMISSIONS))})",
|
||||
)
|
||||
pc.add_argument("--password", help="访问密码 (可选)")
|
||||
pc.add_argument("--ttl-seconds", type=int, help="有效期(秒), 与 --ttl-days 互斥")
|
||||
pc.add_argument("--ttl-days", type=int, help="有效期(天), 与 --ttl-seconds 互斥")
|
||||
pc.add_argument("--length", type=int, default=6, help="短码长度 (4-16, 默认 6)")
|
||||
pc.add_argument("--note", help="备注")
|
||||
pc.set_defaults(func=cmd_create)
|
||||
|
||||
# ---- resolve ----
|
||||
pr = sub.add_parser("resolve", help="解析短链接")
|
||||
pr.add_argument("code", help="短码, 如 ABC123")
|
||||
pr.add_argument("--password", help="访问密码")
|
||||
pr.add_argument(
|
||||
"--dry-run", action="store_true",
|
||||
help="不记录访问 (查询但不计数)",
|
||||
)
|
||||
pr.set_defaults(func=cmd_resolve)
|
||||
|
||||
# ---- revoke ----
|
||||
px = sub.add_parser("revoke", help="撤销短链接")
|
||||
px.add_argument("code", help="短码")
|
||||
px.add_argument("--owner", help="owner 校验 (可选, 不传则不强校)")
|
||||
px.set_defaults(func=cmd_revoke)
|
||||
|
||||
# ---- list ----
|
||||
pl = sub.add_parser("list", help="列出某报告/某用户的链接")
|
||||
pl.add_argument("--report", help="按 report_id 过滤")
|
||||
pl.add_argument("--owner", help="按 owner_id 过滤")
|
||||
pl.add_argument("--limit", type=int, default=100, help="最多返回条数")
|
||||
pl.set_defaults(func=cmd_list)
|
||||
|
||||
# ---- stats ----
|
||||
ps = sub.add_parser("stats", help="查看短链接统计")
|
||||
ps.add_argument("code", help="短码")
|
||||
ps.set_defaults(func=cmd_stats)
|
||||
|
||||
# ---- purge ----
|
||||
pp = sub.add_parser("purge", help="清理过期记录")
|
||||
pp.set_defaults(func=cmd_purge)
|
||||
|
||||
return p
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
parser = build_parser()
|
||||
args = parser.parse_args(argv)
|
||||
return args.func(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user