Release 20250502

.env.sample

@@ -3,3 +3,5 @@ DATABASE_HOST="127.0.0.1"
 DATABASE_PORT=3306
 DATABASE_USER=""
 DATABASE_PASSWD=""
+
+STATIC_APP_DIR="/app/static"

main.py

@@ -1,17 +1,24 @@
 from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
 
+from src.config import settings
+
 from src.anno import router as anno_router
-from src.plan import router as plan_router
+from src.plan.summary import router as plan_router
+from src.plan.details import router as plan_details_router
 from src.health_check import router as health_check_router
 from src.project import router as project_router
+from src.ICP import router as icp_router
 
 app = FastAPI()
 
-app.mount("/static", StaticFiles(directory="/app/static"), name="static")
+if settings.static_app_dir:
+    app.mount("/static", StaticFiles(directory=settings.static_app_dir), name="static")
 
 app.include_router(anno_router)
 app.include_router(plan_router)
+app.include_router(plan_details_router)
 app.include_router(health_check_router)
 app.include_router(project_router)
+app.include_router(icp_router)
 

src/ICP/__init__.py

@@ -0,0 +1,32 @@
+from fastapi import APIRouter, Request
+from loguru import logger
+
+from src.database import ICP
+
+router = APIRouter()
+
+AllICP = ICP.select()
+
+
+@router.get("/icp")
+async def query_icp(domain: str):
+    icp = next(
+        (icp for icp in AllICP if icp.domain in domain),
+        None,
+    )
+
+    if icp is None:
+        logger.error(f"domain not found: {domain}")
+        return {
+            "domain": domain,
+            "icp_beian": "",
+            "icp_url": "",
+            "icp_entity": "",
+        }
+
+    return {
+        "domain": domain,
+        "icp_beian": icp.beian,
+        "icp_url": icp.url,
+        "icp_entity": icp.entity,
+    }

src/config/__init__.py

@@ -7,6 +7,8 @@ class Settings(BaseSettings):
     database_port: int
     database_user: str
     database_passwd: str
+
+    static_app_dir: str
 
     class Config:
         env_file = ".env"

src/database/__init__.py

@@ -46,11 +46,15 @@ class Meta:
 
 class Plan(Model):
     plan_index = IntegerField()
-    platform = CharField()
-    plan_id = CharField()
     type_id = CharField()
+    plan_id = CharField()
+    title = TextField()
+    price = CharField()
+    original_price = CharField()
     popular = IntegerField()
     available = BooleanField()
+    afdian_id = CharField()
+    yimapay_id = CharField()
 
     class Meta:
         database = db
@@ -72,6 +76,17 @@ class Meta:
         database = db
         table_name = "project"
 
+class ICP(Model):
+    domain = CharField()
+    beian = CharField()
+    entity = CharField()
+    url = CharField()
+
+    class Meta:
+        database = db
+        table_name = "icp"
+
 Anno.create_table()
 Plan.create_table()
-Project.create_table()
+Project.create_table()
+ICP.create_table()

src/plan/__init__.py

@@ -1,55 +0,0 @@
-from datetime import datetime
-from fastapi import APIRouter, Request
-from loguru import logger
-from time import time
-
-from src.database import Plan
-
-router = APIRouter()
-
-CacheExpiration = 60  # 秒
-plan_cache = None
-
-
-@router.get("/plan")
-async def query_anno(type_id: str = "GameTools"):
-    logger.debug(f"type_id: {type_id}")
-
-    if not type_id:
-        logger.error(f"type_id is required")
-        return {"ec": 400, "msg": "type_id is required"}
-
-    now = time()
-    global plan_cache
-    if not plan_cache or (now - plan_cache[1] > CacheExpiration):
-        plan_cache = (Plan.select().where(Plan.available == True).order_by(Plan.plan_index), now)
-
-    data = {
-        "home": [],
-        "more": [],
-    }
-
-    for p in plan_cache[0]:
-        if p.type_id == type_id:
-            data["home"].append(
-                {
-                    "platform": p.platform,
-                    "plan_id": p.plan_id,
-                    "type_id": p.type_id,
-                    "popular": p.popular,
-                }
-            )
-        else:
-            data["more"].append(
-                {
-                    "platform": p.platform,
-                    "plan_id": p.plan_id,
-                    "type_id": p.type_id,
-                    "popular": p.popular,
-                }
-            )
-
-    if not data["home"]:
-        data["home"], data["more"] = data["more"], data["home"]
-
-    return {"ec": 200, "data": data}

src/plan/cache.py

@@ -0,0 +1,14 @@
+from time import time
+from src.database import Plan
+
+CacheExpiration = 60  # 秒
+_plan_cache = None
+
+def get_plan_cache():
+    now = time()
+
+    global _plan_cache
+    if not _plan_cache or (now - _plan_cache[1] > CacheExpiration):
+        _plan_cache = (Plan.select().order_by(Plan.plan_index), now)
+
+    return _plan_cache[0]

src/plan/details.py

@@ -0,0 +1,43 @@
+from fastapi import APIRouter
+from loguru import logger
+
+from .cache import get_plan_cache
+
+router = APIRouter()
+
+
+@router.get("/plan/{plan_id}")
+async def query_details(plan_id: str):
+    logger.debug(f"plan_id: {plan_id}")
+
+    if not plan_id:
+        logger.error(f"plan_id is required")
+        return {"ec": 400, "msg": "plan_id is required"}
+
+    p = next((p for p in get_plan_cache() if p.plan_id == plan_id), None)
+    if not p:
+        logger.error(f"plan_id not found")
+        return {"ec": 404, "msg": "plan_id not found"}
+
+
+    afdian_info = [s.strip() for s in p.afdian_id.split(",")]
+    if len(afdian_info) != 2:
+        logger.error(f"afdian_info is invalid")
+        return {"ec": 500, "msg": "afdian_info is invalid"}
+
+    return {
+        "ec": 200,
+        "data": {
+            "title": p.title,
+            "type_id": p.type_id,
+            "price": p.price,
+            "original_price": p.original_price,
+            "popular": p.popular,
+            # "afdian_id": p.afdian_id,
+            "afdian_info": {
+                "plan_id": afdian_info[0],
+                "sku_id": afdian_info[1],
+            },
+            "yimapay_id": p.yimapay_id,
+        },
+    }

src/plan/summary.py

@@ -0,0 +1,50 @@
+from fastapi import APIRouter
+from loguru import logger
+from time import time
+
+from .cache import get_plan_cache
+
+router = APIRouter()
+
+@router.get("/plan")
+async def query_plan(type_id: str = "GameTools"):
+    logger.debug(f"type_id: {type_id}")
+
+    if not type_id:
+        logger.error(f"type_id is required")
+        return {"ec": 400, "msg": "type_id is required"}
+
+    data = {
+        "home": [],
+        "more": [],
+    }
+
+    for p in get_plan_cache():
+        if not p.available:
+            continue
+
+        if p.type_id == type_id:
+            data["home"].append(
+                {
+                    "title": p.title,
+                    "price": p.price,
+                    "original_price": p.original_price,
+                    "popular": p.popular,
+                    "plan_id": p.plan_id,
+                }
+            )
+        else:
+            data["more"].append(
+                {
+                    "title": p.title,
+                    "price": p.price,
+                    "original_price": p.original_price,
+                    "popular": p.popular,
+                    "plan_id": p.plan_id,
+                }
+            )
+
+    if not data["home"]:
+        data["home"], data["more"] = data["more"], data["home"]
+
+    return {"ec": 200, "data": data}