merge main

2026-06-17 15:45:26 +08:00
parent 0d647a7171 189aef6a82
commit 4114ff1c16
15 changed files with 24258 additions and 8 deletions
@@ -2,4 +2,4 @@
 VITE_API_BASE=http://ceshi.apiforeign.minzhong.cn
 VITE_USE_MOCK=false
 # Encryption disabled for dev — enable in production
-VITE_ENCRYPTION_KEY=
+VITE_ENCRYPTION_KEY=
@@ -0,0 +1,5 @@
 #VITE_API_BASE=http://localhost:8000
 VITE_API_BASE=http://apiforeign.minzhong.cn
 VITE_USE_MOCK=false
 # Encryption disabled for dev — enable in production
 VITE_ENCRYPTION_KEY=
@@ -28,7 +28,7 @@
        }
      })();
    </script>
-    <script type="module" crossorigin src="/assets/index-CDdTfqKl.js"></script>
+    <script type="module" crossorigin src="/assets/index-CLhvZPpk.js"></script>
    <link rel="stylesheet" crossorigin href="/assets/index-D7ShJUt4.css">
  </head>
  <body>
@@ -21,6 +21,7 @@ from app.api.v1.test import router as test_router
 from app.api.v1.user_oauth import router as user_oauth_router
 from app.api.v1.user_oauth_app import router as user_oauth_app_router
 from app.api.v1.upload_material import router as upload_material_router
 from app.api.v1.pre_test_template import router as pre_test_template_router
 api_router = APIRouter()
 api_router.include_router(auth_router)
@@ -44,3 +45,4 @@ api_router.include_router(test_router)
 api_router.include_router(user_oauth_router)
 api_router.include_router(user_oauth_app_router)
 api_router.include_router(upload_material_router)
 api_router.include_router(pre_test_template_router)
@@ -0,0 +1,358 @@
 import json
 from typing import Any, Optional, Dict
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select
 from app.models.pre_test_template import PreTestTemplate
 from app.utils.douyinApi import DouyinApi
 from app.utils.area import parse_district_data, get_area_by_level, get_cached_area_data, fetch_and_cache_area_data
 from app.dependencies import get_current_user, get_db
 from app.models.user import User
 from app.schemas.pre_test_template import (
    PreTestTemplateCreate,
    PreTestTemplateListResponse,
    PreTestTemplateOut,
    PreTestTemplateUpdate,
 )
 from app.services.pre_test_template_service import (
    create_pre_test_template,
    delete_pre_test_template,
    get_default_template,
    get_pre_test_template,
    get_pre_test_template_list,
    update_pre_test_template,
 )
 router = APIRouter(prefix="/pre-test-template", tags=["前测模板"])
 def _template_to_dict(template):
    return {
        "id": template.id,
        "name": template.name,
        "note": template.note,
        "platform": template.platform,
        "external_action": template.external_action,
        "cpa_bid": template.cpa_bid,
        "audience_gender": template.audience_gender,
        "audience_age": json.loads(template.audience_age) if template.audience_age else None,
        "audience_region": json.loads(template.audience_region) if template.audience_region else None,
        "audience_network": json.loads(template.audience_network) if template.audience_network else None,
        "cus_name": template.cus_name,
        "pricing_type": template.pricing_type,
        "cost_cap": template.cost_cap,
        "target_cost": template.target_cost,
        "nobid": template.nobid,
        "cpc_bid": template.cpc_bid,
        "budget": template.budget,
        "is_default": template.is_default,
        "user_id": template.user_id,
        "created_at": template.created_at,
        "updated_at": template.updated_at,
    }
@router.post(
    "/create",
    summary="创建前测模板",
    description="创建一个新的前测模板",
 )
 async def create_template(
    req: PreTestTemplateCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
 ) -> Any:
    try:
        #新增一个判断，如果模板名称重复，提示用户修改
        result = await db.execute(
            select(PreTestTemplate).where(
                PreTestTemplate.name == req.name,
                PreTestTemplate.user_id == current_user.id,
                PreTestTemplate.deleted_at.is_(None),
            )
        )
        existing_template = result.scalar()
        if existing_template:
            raise ValueError("模板名称已存在")
        template = await create_pre_test_template(
            user_id=current_user.id,
            db=db,
            name=req.name,
            note=req.note,
            platform=req.platform,
            external_action=req.external_action,
            cpa_bid=req.cpa_bid,
            audience_gender=req.audience_gender,
            audience_age=req.audience_age,
            audience_region=req.audience_region,
            audience_network=req.audience_network,
            cus_name=req.cus_name,
            pricing_type=req.pricing_type,
            cost_cap=req.cost_cap,
            target_cost=req.target_cost,
            nobid=req.nobid,
            cpc_bid=req.cpc_bid,
            budget=req.budget,
            is_default=req.is_default,
        )
        return {
            "code": 0,
            "message": "创建成功",
        }
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"创建失败: {str(e)}",
        )
@router.get(
    "/list",
    summary="获取前测模板列表",
    description="获取当前用户的前测模板列表，支持按平台筛选和分页",
    response_model=PreTestTemplateListResponse,
 )
 async def list_templates(
    platform: Optional[str] = Query(None, description="投放平台筛选（AD/QIANCHUAN/LOCAL）"),
    page: int = Query(1, description="页码，默认1"),
    page_size: int = Query(10, description="每页数量，默认10，最大100"),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
 ) -> Any:
    try:
        result = await get_pre_test_template_list(
            user_id=current_user.id,
            db=db,
            platform=platform,
            page=page,
            page_size=page_size,
        )
        return {
            "code": 0,
            "message": "查询成功",
            "data": [_template_to_dict(t) for t in result["data"]],
            "pagination": {
                "page": result["page"],
                "page_size": result["page_size"],
                "total": result["total"],
                "total_pages": (result["total"] + result["page_size"] - 1) // result["page_size"],
            },
        }
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询失败: {str(e)}",
        )
@router.get(
    "/default",
    summary="获取默认前测模板",
    description="获取当前用户的默认前测模板",
 )
 async def get_default(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
 ) -> Any:
    try:
        template = await get_default_template(current_user.id, db)
        if not template:
            return {
                "code": 0,
                "message": "未设置默认模板",
                "data": None,
            }
        return {
            "code": 0,
            "message": "查询成功",
            "data": _template_to_dict(template),
        }
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询失败: {str(e)}",
        )
@router.get(
    "/select/{template_id}",
    summary="获取前测模板详情",
    description="根据模板id获取前测模板详情",
 )
 async def get_template(
    template_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
 ) -> Any:
    try:
        template = await get_pre_test_template(template_id, current_user.id, db)
        if not template:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="模板不存在",
            )
        return {
            "code": 0,
            "message": "查询成功",
            "data": _template_to_dict(template),
        }
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询失败: {str(e)}",
        )
@router.post(
    "/update/{template_id}",
    summary="更新前测模板",
    description="更新指定的前测模板",
 )
 async def update_template(
    template_id: str,
    req: PreTestTemplateUpdate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
 ) -> Any:
    try:
        update_data = req.dict(exclude_none=True)
        template = await update_pre_test_template(template_id, current_user.id, db, **update_data)
        if not template:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="模板不存在",
            )
        return {
            "code": 0,
            "message": "更新成功",
            "data": _template_to_dict(template),
        }
    except HTTPException:
        raise
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"更新失败: {str(e)}",
        )
@router.get(
    "/delete/{template_id}",
    summary="删除前测模板",
    description="软删除指定的前测模板",
 )
 async def delete_template(
    template_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
 ) -> Any:
    try:
        success = await delete_pre_test_template(template_id, current_user.id, db)
        if not success:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="模板不存在",
            )
        return {
            "code": 0,
            "message": "删除成功",
        }
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"删除失败: {str(e)}",
        )
@router.get(
    "/getArea",
    summary="获取行政区域信息",
    description="获取指定级别的行政区域信息，支持一级、二级、三级区域,如果需要更新地区，执行：/api/pre-test-template/getArea?oauth_id=0019ecab9b8bc57d964&advertiser_id=1836693172153543",
 )
 async def get_template_area(
    oauth_id: str = Query(None, description="授权ID选填，更新地区必填"),
    advertiser_id: str = Query(default="1836693172153543", description="授权ID选填，更新地区必填"),
    code: Optional[str] = Query("CN", description="行政区域编码，默认中国CN，选填"),
    level: Optional[str] = Query("ONE_LEVEL", description="行政区域层级，可选值：ONE_LEVEL(获取省份)、TWO_LEVEL（市级）、THREE_LEVEL（区级）"),
    parent_code: Optional[str] = Query(None, description="父级区域编码，获取二级时传一级编码，获取三级时传二级编码"),
 ) -> Any:
    try:
        # 1. 先检查缓存是否存在
        area_list = get_cached_area_data()
        # 2. 如果缓存不存在，调用接口获取数据并保存到缓存
        if not area_list:
            area_list = await fetch_and_cache_area_data(oauth_id, advertiser_id, code)
        # 3. 根据 level 参数过滤区域
        if level == "ONE_LEVEL":
            result = get_area_by_level(area_list, "ONE_LEVEL")
        elif level == "TWO_LEVEL":
            result = get_area_by_level(area_list, "TWO_LEVEL", parent_code)
        elif level == "THREE_LEVEL":
            if not parent_code:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail="获取三级区域需要提供二级区域编码(parent_code)",
                )
            result = get_area_by_level(area_list, "THREE_LEVEL", parent_code)
        else:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"不支持的级别: {level}",
            )
        # 4. 转换为字典格式返回
        result_dict = [area.to_dict() for area in result]
        return {
            "code": 0,
            "message": "成功",
            "data": result_dict,
        }
    except HTTPException:
        raise
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取区域信息失败: {str(e)}",
        )
@@ -123,7 +123,7 @@ async def send_sms_code(
    description="校验指定手机号、场景下的短信验证码。业务接口一般会内部校验，本接口主要用于前端调试或单独校验。",
 )
 async def verify_sms(req: SmsVerifyRequest):
-    ok = await verify_sms_code(req.phone, req.code, req.scene.value)
+    ok = await verify_sms_code(req.phone.strip(), req.code.strip(), req.scene.value)
    if not ok:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
@@ -6,13 +6,12 @@ from app.dependencies import get_current_user, get_db
 from app.models.user import User
 from app.models.user_oauth import UserOAuth
 from app.models.user_oauth_app import UserOAuthApp
-from app.schemas.user_oauth import RequestOAuthRequest, RequestOAuthResponse, UserOAuthOut
+from app.schemas.user_oauth import RequestOAuthRequest, RequestOAuthResponse, UserOAuthOut, OAuthListResponse
 from app.services.user_oauth_service import (
    build_oauth_url,
    get_token,
    get_oauth_list,
 )
 from app.tasks.user_oauth_tasks import _update_oauth_accounts
 router = APIRouter(prefix="/user-oauth", tags=["oauth"])
@@ -108,6 +107,7 @@ async def juliang_callback(
    "/oauth_list",
    summary="获取账户下所有授权列表",
    description="获取当前用户下所有授权账户列表，支持按授权登录账号、开户方式、授权账户id筛选",
    response_model=OAuthListResponse,
 )
 async def oauth_list(
    account_userid: str | None = Query(None, description="授权登录账号id"),
@@ -0,0 +1,111 @@
 from datetime import datetime
 from typing import List, Optional
 from pydantic import BaseModel, Field, field_validator
 class PreTestTemplateCreate(BaseModel):
    name: str = Field(..., description="模板名称")
    note: Optional[str] = Field(None, description="模板备注")
    platform: Optional[str] = Field(None, description="投放平台（AD/QIANCHUAN/LOCAL）")
    external_action: Optional[str] = Field(None, description="转化目标")
    cpa_bid: Optional[float] = Field(None, description="目标转化成本：[1, 10000]")
    audience_gender: Optional[str] = Field(None, description="性别（ALL/MALE/FEMALE）")
    audience_age: Optional[List[str]] = Field(None, description="受众年龄列表")
    audience_region: Optional[List[int]] = Field(None, description="受众地区code列表")
    audience_network: Optional[List[str]] = Field(None, description="网络类型列表")
    cus_name: Optional[str] = Field(None, description="客户主体名称")
    pricing_type: Optional[str] = Field(None, description="出价类型（OCPC/CPA/OCPM）")
    cost_cap: Optional[bool] = Field(None, description="是否最优成本出价（仅AD支持）")
    target_cost: Optional[bool] = Field(None, description="是否稳定成本出价（仅AD支持）")
    nobid: Optional[bool] = Field(None, description="是否最大转化出价（仅AD支持）")
    cpc_bid: Optional[float] = Field(None, description="目标点击成本：[1, 10000]")
    budget: Optional[float] = Field(None, description="预算金额：[1, 10000]")
    is_default: Optional[bool] = Field(None, description="是否设为默认模板")
    @field_validator('platform')
    def validate_platform(cls, v):
        if v is None:
            return v
        allowed = ["AD", "QIANCHUAN", "LOCAL"]
        if v not in allowed:
            raise ValueError(f"platform must be one of {allowed}")
        return v
    @field_validator('pricing_type')
    def validate_pricing_type(cls, v):
        if v is None:
            return v
        allowed = ["OCPC", "CPA", "OCPM"]
        if v not in allowed:
            raise ValueError(f"pricing_type must be one of {allowed}")
        return v
    @field_validator('audience_gender')
    def validate_gender(cls, v):
        if v is None:
            return v
        allowed = ["ALL", "MALE", "FEMALE"]
        if v not in allowed:
            raise ValueError(f"audience_gender must be one of {allowed}")
        return v
 class PreTestTemplateUpdate(BaseModel):
    name: Optional[str] = Field(None, description="模板名称")
    note: Optional[str] = Field(None, description="模板备注")
    platform: Optional[str] = Field(None, description="投放平台")
    external_action: Optional[str] = Field(None, description="转化目标")
    cpa_bid: Optional[float] = Field(None, description="目标转化成本")
    audience_gender: Optional[str] = Field(None, description="性别")
    audience_age: Optional[List[str]] = Field(None, description="受众年龄列表")
    audience_region: Optional[List[int]] = Field(None, description="受众地区code列表")
    audience_network: Optional[List[str]] = Field(None, description="网络类型列表")
    cus_name: Optional[str] = Field(None, description="客户主体名称")
    pricing_type: Optional[str] = Field(None, description="出价类型")
    cost_cap: Optional[bool] = Field(None, description="是否最优成本出价")
    target_cost: Optional[bool] = Field(None, description="是否稳定成本出价")
    nobid: Optional[bool] = Field(None, description="是否最大转化出价")
    cpc_bid: Optional[float] = Field(None, description="目标点击成本")
    budget: Optional[float] = Field(None, description="预算金额")
    is_default: Optional[bool] = Field(None, description="是否设为默认模板")
 class PreTestTemplateOut(BaseModel):
    id: str = Field(..., description="主键")
    name: str = Field(..., description="模板名称")
    note: Optional[str] = Field(None, description="模板备注")
    platform: Optional[str] = Field(None, description="投放平台")
    external_action: Optional[str] = Field(None, description="转化目标")
    cpa_bid: Optional[float] = Field(None, description="目标转化成本")
    audience_gender: Optional[str] = Field(None, description="性别")
    audience_age: Optional[List[str]] = Field(None, description="受众年龄列表")
    audience_region: Optional[List[int]] = Field(None, description="受众地区code列表")
    audience_network: Optional[List[str]] = Field(None, description="网络类型列表")
    cus_name: Optional[str] = Field(None, description="客户主体名称")
    pricing_type: Optional[str] = Field(None, description="出价类型")
    cost_cap: Optional[bool] = Field(None, description="是否最优成本出价")
    target_cost: Optional[bool] = Field(None, description="是否稳定成本出价")
    nobid: Optional[bool] = Field(None, description="是否最大转化出价")
    cpc_bid: Optional[float] = Field(None, description="目标点击成本")
    budget: Optional[float] = Field(None, description="预算金额")
    is_default: Optional[bool] = Field(None, description="是否默认模板")
    user_id: str = Field(..., description="用户id")
    created_at: datetime = Field(..., description="创建时间")
    updated_at: datetime = Field(..., description="更新时间")
    model_config = {"from_attributes": True}
 class PaginationInfo(BaseModel):
    page: int = Field(..., description="当前页码")
    page_size: int = Field(..., description="每页数量")
    total: int = Field(..., description="总记录数")
    total_pages: int = Field(..., description="总页数")
 class PreTestTemplateListResponse(BaseModel):
    code: int = Field(0, description="返回码")
    message: str = Field("查询成功", description="返回消息")
    data: List[PreTestTemplateOut] = Field(..., description="模板列表")
    pagination: PaginationInfo = Field(..., description="分页信息")
@@ -0,0 +1,258 @@
 import json
 from typing import Optional
 from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.models.pre_test_template import PreTestTemplate
 from app.utils.id_gen import generate_id
 async def create_pre_test_template(
    user_id: str,
    db: AsyncSession,
    name: str,
    note: Optional[str] = None,
    platform: Optional[str] = None,
    external_action: Optional[str] = None,
    cpa_bid: Optional[float] = None,
    audience_gender: Optional[str] = None,
    audience_age: Optional[list] = None,
    audience_region: Optional[list] = None,
    audience_network: Optional[list] = None,
    cus_name: Optional[str] = None,
    pricing_type: Optional[str] = None,
    cost_cap: Optional[bool] = None,
    target_cost: Optional[bool] = None,
    nobid: Optional[bool] = None,
    cpc_bid: Optional[float] = None,
    budget: Optional[float] = None,
    is_default: Optional[bool] = None,
 ) -> PreTestTemplate:
    if is_default:
        await db.execute(
            update(PreTestTemplate)
            .where(
                PreTestTemplate.user_id == user_id,
                PreTestTemplate.is_default == True,
                PreTestTemplate.deleted_at.is_(None),
            )
            .values(is_default=False)
        )
    template = PreTestTemplate(
        id=generate_id(),
        user_id=user_id,
        name=name,
        note=note,
        platform=platform,
        external_action=external_action,
        cpa_bid=cpa_bid,
        audience_gender=audience_gender,
        audience_age=json.dumps(audience_age) if audience_age else None,
        audience_region=json.dumps(audience_region) if audience_region else None,
        audience_network=json.dumps(audience_network) if audience_network else None,
        cus_name=cus_name,
        pricing_type=pricing_type,
        cost_cap=cost_cap,
        target_cost=target_cost,
        nobid=nobid,
        cpc_bid=cpc_bid,
        budget=budget,
        is_default=is_default,
    )
    db.add(template)
    await db.commit()
    await db.refresh(template)
    return template
 async def update_pre_test_template(
    template_id: str,
    user_id: str,
    db: AsyncSession,
    **kwargs,
 ) -> Optional[PreTestTemplate]:
    result = await db.execute(
        select(PreTestTemplate).where(
            PreTestTemplate.id == template_id,
            PreTestTemplate.user_id == user_id,
            PreTestTemplate.deleted_at.is_(None),
        )
    )
    template = result.scalar_one_or_none()
    if not template:
        return None
    if kwargs.get("is_default"):
        await db.execute(
            update(PreTestTemplate)
            .where(
                PreTestTemplate.user_id == user_id,
                PreTestTemplate.is_default == True,
                PreTestTemplate.id != template_id,
                PreTestTemplate.deleted_at.is_(None),
            )
            .values(is_default=False)
        )
    update_data = {}
    if "name" in kwargs:
        update_data["name"] = kwargs["name"]
    if "note" in kwargs:
        update_data["note"] = kwargs["note"]
    if "platform" in kwargs:
        update_data["platform"] = kwargs["platform"]
    if "external_action" in kwargs:
        update_data["external_action"] = kwargs["external_action"]
    if "cpa_bid" in kwargs:
        update_data["cpa_bid"] = kwargs["cpa_bid"]
    if "audience_gender" in kwargs:
        update_data["audience_gender"] = kwargs["audience_gender"]
    if "audience_age" in kwargs:
        update_data["audience_age"] = json.dumps(kwargs["audience_age"]) if kwargs["audience_age"] else None
    if "audience_region" in kwargs:
        update_data["audience_region"] = json.dumps(kwargs["audience_region"]) if kwargs["audience_region"] else None
    if "audience_network" in kwargs:
        update_data["audience_network"] = json.dumps(kwargs["audience_network"]) if kwargs["audience_network"] else None
    if "cus_name" in kwargs:
        update_data["cus_name"] = kwargs["cus_name"]
    if "pricing_type" in kwargs:
        update_data["pricing_type"] = kwargs["pricing_type"]
    if "cost_cap" in kwargs:
        update_data["cost_cap"] = kwargs["cost_cap"]
    if "target_cost" in kwargs:
        update_data["target_cost"] = kwargs["target_cost"]
    if "nobid" in kwargs:
        update_data["nobid"] = kwargs["nobid"]
    if "cpc_bid" in kwargs:
        update_data["cpc_bid"] = kwargs["cpc_bid"]
    if "budget" in kwargs:
        update_data["budget"] = kwargs["budget"]
    if "is_default" in kwargs:
        update_data["is_default"] = kwargs["is_default"]
    if update_data:
        await db.execute(
            update(PreTestTemplate)
            .where(PreTestTemplate.id == template_id)
            .values(**update_data)
        )
        await db.commit()
        await db.refresh(template)
    return template
 async def delete_pre_test_template(
    template_id: str,
    user_id: str,
    db: AsyncSession,
 ) -> bool:
    result = await db.execute(
        select(PreTestTemplate).where(
            PreTestTemplate.id == template_id,
            PreTestTemplate.user_id == user_id,
            PreTestTemplate.deleted_at.is_(None),
        )
    )
    template = result.scalar_one_or_none()
    if not template:
        return False
    from datetime import datetime
    template.deleted_at = datetime.now()
    await db.commit()
    return True
 async def get_pre_test_template(
    template_id: str,
    user_id: str,
    db: AsyncSession,
 ) -> Optional[PreTestTemplate]:
    result = await db.execute(
        select(PreTestTemplate).where(
            PreTestTemplate.id == template_id,
            PreTestTemplate.user_id == user_id,
            PreTestTemplate.deleted_at.is_(None),
        )
    )
    return result.scalar_one_or_none()
 async def get_pre_test_template_list(
    user_id: str,
    db: AsyncSession,
    platform: Optional[str] = None,
    page: int = 1,
    page_size: int = 10,
 ) -> dict:
    if page < 1:
        page = 1
    if page_size < 1:
        page_size = 10
    if page_size > 100:
        page_size = 100
    query = select(PreTestTemplate).where(
        PreTestTemplate.user_id == user_id,
        PreTestTemplate.deleted_at.is_(None),
    )
    if platform:
        query = query.where(PreTestTemplate.platform == platform)
    query = query.order_by(PreTestTemplate.is_default.desc(), PreTestTemplate.created_at.desc())
    total_result = await db.execute(query.with_only_columns(PreTestTemplate.id))
    total = len(total_result.scalars().all())
    offset = (page - 1) * page_size
    query = query.offset(offset).limit(page_size)
    result = await db.execute(query)
    templates = result.scalars().all()
    return {
        "data": templates,
        "total": total,
        "page": page,
        "page_size": page_size,
    }
 async def get_default_template(
    user_id: str,
    db: AsyncSession,
 ) -> Optional[PreTestTemplate]:
    # 如果有默认模板，返回默认模板
    # 如果有多条默认的模板，返回最新创建的模板
    # 如果没有设置默认模板，返回最新创建的模板
    result = await db.execute(
        select(PreTestTemplate)
        .where(
            PreTestTemplate.user_id == user_id,
            PreTestTemplate.is_default == True,
            PreTestTemplate.deleted_at.is_(None),
        )
        .order_by(PreTestTemplate.created_at.desc())
    )
    template = result.scalar_one_or_none()
    if template:
        return template
    # 如果没有默认模板，返回最新创建的模板
    result = await db.execute(
        select(PreTestTemplate)
        .where(
            PreTestTemplate.user_id == user_id,
            PreTestTemplate.deleted_at.is_(None),
        )
        .order_by(PreTestTemplate.created_at.desc())
    )
    return result.scalar_one_or_none()
@@ -173,7 +173,7 @@ async def verify_sms_code(phone: str, code: str, scene: str = "login") -> bool:
        stored = await redis.get(key)
        if isinstance(stored, bytes):
            stored = stored.decode()
-        if stored and str(stored) == str(code):
+        if stored is not None and stored == code:
            await redis.delete(key)
            return True
        return False
@@ -0,0 +1,241 @@
 import json
 import os
 from typing import Optional, List, Dict
 # 缓存文件路径
 CACHE_FILE_PATH = os.path.join(os.path.dirname(__file__), 'area_cache.json')
 class AreaInfo:
    def __init__(self, code: str, name: str, level: str, geoname_id: int = None):
        self.code = code
        self.name = name
        self.level = level
        self.geoname_id = geoname_id
        self.sub_districts: List[AreaInfo] = []
    def to_dict(self):
        return {
            "code": self.code,
            "name": self.name,
            "level": self.level,
            "geoname_id": self.geoname_id,
            "sub_districts": [sd.to_dict() for sd in self.sub_districts] if self.sub_districts else [],
        }
    @classmethod
    def from_dict(cls, data: dict):
        area = cls(
            code=data.get("code"),
            name=data.get("name"),
            level=data.get("level"),
            geoname_id=data.get("geoname_id"),
        )
        sub_districts = data.get("sub_districts", [])
        for sub in sub_districts:
            area.sub_districts.append(cls.from_dict(sub))
        return area
 def parse_district_data(district_data: dict) -> AreaInfo:
    """解析巨量接口返回的区域数据"""
    area = AreaInfo(
        code=district_data.get("code"),
        name=district_data.get("name"),
        level=district_data.get("level"),
        geoname_id=district_data.get("geoname_id"),
    )
    sub_districts = district_data.get("sub_districts")
    if sub_districts:
        for sub in sub_districts:
            area.sub_districts.append(parse_district_data(sub))
    return area
 def save_area_cache(areas: List[AreaInfo]) -> None:
    """将区域数据保存到缓存文件"""
    data = [area.to_dict() for area in areas]
    with open(CACHE_FILE_PATH, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
 def load_area_cache() -> Optional[List[AreaInfo]]:
    """从缓存文件加载区域数据"""
    if not os.path.exists(CACHE_FILE_PATH):
        return None
    try:
        with open(CACHE_FILE_PATH, 'r', encoding='utf-8') as f:
            data = json.load(f)
            return [AreaInfo.from_dict(item) for item in data]
    except (json.JSONDecodeError, IOError):
        return None
 def clear_area_cache() -> None:
    """清除缓存文件"""
    if os.path.exists(CACHE_FILE_PATH):
        os.remove(CACHE_FILE_PATH)
 def filter_by_level(areas: List[AreaInfo], target_level: str) -> List[AreaInfo]:
    """
    根据级别过滤区域信息
    :param areas: 区域列表
    :param target_level: ONE_LEVEL / TWO_LEVEL / THREE_LEVEL
    :return: 指定级别的区域列表
    """
    result = []
    def traverse(area: AreaInfo):
        if area.level == target_level:
            filtered = AreaInfo(
                code=area.code,
                name=area.name,
                level=area.level,
                geoname_id=area.geoname_id,
            )
            result.append(filtered)
        if area.sub_districts:
            for sub in area.sub_districts:
                traverse(sub)
    for area in areas:
        traverse(area)
    return result
 def get_first_level_areas(areas: List[AreaInfo]) -> List[AreaInfo]:
    """获取所有一级区域（省/直辖市）"""
    return filter_by_level(areas, "ONE_LEVEL")
 def get_second_level_areas(areas: List[AreaInfo], parent_code: str = None) -> List[AreaInfo]:
    """
    获取二级区域（市）
    :param areas: 区域列表
    :param parent_code: 一级区域code，不传则返回所有二级区域
    :return: 二级区域列表
    """
    if parent_code:
        def find_parent_and_get_children(area: AreaInfo):
            if area.code == parent_code:
                return [AreaInfo(
                    code=sub.code,
                    name=sub.name,
                    level=sub.level,
                    geoname_id=sub.geoname_id,
                ) for sub in area.sub_districts] if area.sub_districts else []
            if area.sub_districts:
                for sub in area.sub_districts:
                    result = find_parent_and_get_children(sub)
                    if result:
                        return result
            return []
        for area in areas:
            result = find_parent_and_get_children(area)
            if result:
                return result
        return []
    else:
        return filter_by_level(areas, "TWO_LEVEL")
 def get_third_level_areas(areas: List[AreaInfo], parent_code: str) -> List[AreaInfo]:
    """
    获取三级区域（区/县）
    :param areas: 区域列表
    :param parent_code: 二级区域code
    :return: 三级区域列表
    """
    def find_parent_and_get_children(area: AreaInfo):
        if area.code == parent_code:
            return [AreaInfo(
                code=sub.code,
                name=sub.name,
                level=sub.level,
                geoname_id=sub.geoname_id,
            ) for sub in area.sub_districts] if area.sub_districts else []
        if area.sub_districts:
            for sub in area.sub_districts:
                result = find_parent_and_get_children(sub)
                if result:
                    return result
        return []
    for area in areas:
        result = find_parent_and_get_children(area)
        if result:
            return result
    return []
 def get_area_by_level(areas: List[AreaInfo], level: str, parent_code: str = None) -> List[AreaInfo]:
    """
    根据级别获取区域信息
    :param areas: 区域列表
    :param level: ONE_LEVEL / TWO_LEVEL / THREE_LEVEL
    :param parent_code: 父级区域code（TWO_LEVEL和THREE_LEVEL时可选/必填）
    :return: 区域列表
    """
    level = level.upper()
    if level == "ONE_LEVEL":
        return get_first_level_areas(areas)
    elif level == "TWO_LEVEL":
        return get_second_level_areas(areas, parent_code)
    elif level == "THREE_LEVEL":
        if not parent_code:
            raise ValueError("获取三级区域需要提供二级区域code")
        return get_third_level_areas(areas, parent_code)
    else:
        raise ValueError(f"不支持的级别: {level}")
 async def fetch_and_cache_area_data(oauth_id: str, advertiser_id: str = 1836693172153543, code: str = "CN") -> List[AreaInfo]:
    """
    从接口获取区域数据并缓存到文件
    :param oauth_id: 授权ID
    :param code: 行政区域编码，默认中国CN
    :return: 区域列表
    """
    from app.utils.douyinApi import DouyinApi
    params = {
        "advertiser_id": advertiser_id,
        "codes": json.dumps([code]),
        "language": "ZH_CN",
        "sub_district": "THREE_LEVEL",
        "version": "V2_3_2"
    }
    area_response = await DouyinApi().get_area(oauth_id=oauth_id, params=params)
    if area_response.get("code") != 0:
        raise Exception(f"获取区域信息失败: {area_response.get('message', '未知错误')}")
    districts_data = area_response.get("data", {}).get("districts", [])
    if not districts_data:
        raise Exception("接口返回的区域数据为空")
    area_list = [parse_district_data(d) for d in districts_data]
    save_area_cache(area_list)
    return area_list
 def get_cached_area_data() -> Optional[List[AreaInfo]]:
    """获取缓存的区域数据"""
    return load_area_cache()
@@ -52,4 +52,17 @@ class DouyinApi:
            url,
            'POST',
            options
        )
    #获取区域信息
    async def get_area(self, oauth_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        if not oauth_id:
            raise RuntimeError('OAuth ID is not set.')
        url = "https://api.oceanengine.com/open_api/2/tools/admin/info/"
        return await self.request.request_with_token_with_context(
            oauth_id,
            url,
            'GET',
            {'params': params or {}}
        )
@@ -0,0 +1,5 @@
 #VITE_API_BASE=http://localhost:8000
 VITE_API_BASE=http://apiforeign.minzhong.cn
 VITE_USE_MOCK=false
 # Encryption disabled for dev — enable in production
 VITE_ENCRYPTION_KEY=