deploy: bbf5721

.dockerignore

@@ -0,0 +1,23 @@
+.git
+.github
+.venv
+.vscode
+.claude
+.pytest_cache
+__pycache__
+**/__pycache__
+*.py[oc]
+
+.env
+.env.*
+**/.env
+**/.env.*
+scam-detector-ocr-*.json
+credentials.json
+
+tests
+scripts
+backend/models
+data/chroma
+
+AGENTS.md

.gitattributes

@@ -0,0 +1 @@
+README.md merge=ours

.github/workflows/check_size.yaml

@@ -0,0 +1,16 @@
+name: Check file size
+on:               # or directly `on: [push]` to run the action on every push on any branch
+  pull_request:
+    branches: [space-deploy]
+
+  # to run this workflow manually from the Actions tab
+  workflow_dispatch:
+
+jobs:
+  sync-to-hub:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check large files
+        uses: ActionsDesk/lfs-warning@v2.0
+        with:
+          filesizelimit: 10485760 # this is 10MB so we can sync to HF Spaces

.github/workflows/sync_to_hf.yaml

@@ -0,0 +1,28 @@
+name: Sync to Hugging Face hub
+on:
+  push:
+    branches: [space-deploy]
+
+  # to run this workflow manually from the Actions tab
+  workflow_dispatch:
+
+jobs:
+  sync-to-hub:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          fetch-depth: 1
+          lfs: true
+      # 只推送不含歷史的快照,避免 tests/ 內的二進位檔(及其歷史)被 HF 拒絕
+      - name: Push to hub
+        env:
+          HF_TOKEN: ${{ secrets.HF_TOKEN }}
+        run: |
+          git config user.name "github-actions"
+          git config user.email "github-actions@users.noreply.github.com"
+          git checkout --orphan hf-snapshot
+          git rm -r -q --cached --ignore-unmatch tests
+          git commit -q -m "deploy: ${GITHUB_SHA::7}"
+          git remote add hf https://kko12:$HF_TOKEN@huggingface.co/spaces/kko12/spam-detector
+          git push -f hf hf-snapshot:main

.gitignore

@@ -0,0 +1,26 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+.vscode
+
+backend/models
+.env
+.env.*
+backend/.env
+backend/.env.*
+!.env.example
+
+data/
+db/
+
+dev.*
+
+scam-detector-ocr-*.json
+credentials.json

.python-version

@@ -0,0 +1 @@
+3.14

AGENTS.md

@@ -0,0 +1,77 @@
+## Code style guidelines
+ 
+### 1. 實作前先產生計畫
+ 
+在進行任何非瑣碎的程式碼變更前(新增功能、重構、涉及多個檔案的 bug 修復),
+先以下列格式產出計畫,再開始實作:
+ 
+1. **目標**:一句話描述要達成什麼
+2. **影響範圍**:會修改／新增哪些檔案或模組
+3. **拆分步驟**:條列式列出實作順序(每一步應是可獨立驗證的最小單位)
+4. **風險／取捨**:是否有需要注意的邊界情況、相容性、效能影響
+5. **驗證方式**:完成後如何確認正確(跑哪個測試、哪個指令)
+計畫確認後才開始寫程式碼。若使用者未回覆確認,預設視為同意;
+但遇到以下情況,必須停下來等待明確確認才可繼續:
+ 
+- 會刪除或覆寫現有邏輯
+- 影響資料庫 schema 或外部 API 合約
+- 實作方式有多種合理選項,需要使用者取捨判斷
+單一函式修正、拼字／格式修正等瑣碎變更可略過此流程,直接實作。
+ 
+### 2. 巢狀與職責集中度
+ 
+- 函式巢狀深度盡量不超過 2 層;超過時優先用 early return／guard clause 打平,
+  而非直接視為「必須拆函式」的訊號
+- 函式長度沒有硬性上限;只有當同一函式混雜多種職責
+  (例如同時做 I/O 呼叫、資料轉換、商業邏輯判斷)時才需要拆分
+- 一個檔案／模組只負責單一領域(例如「搜尋」「總結」「快取」分開放)
+
+### 3. 函式拆分準則(避免過度拆分)
+ 
+只有符合以下任一條件,才拆成獨立函式:
+ 
+1. 邏輯會被複用 ≥ 2 次
+2. 該段邏輯是獨立語意單元,可脫離上下文理解,適合單獨寫測試
+3. 屬於不同關注點(I/O ／純運算／格式化／驗證)
+不要拆分的情況:
+ 
+- 只執行一次、線性依序執行的步驟,硬切成多個小函式
+- 拆分後函式間需要傳遞大量參數或共享可變狀態(代表本質上是同一件事)
+- 函式名稱與內容幾乎一比一翻譯,沒有抽象價值(例如 `add_one(x): return x + 1`)
+拆分的目的是讓呼叫端讀起來像文件,而不是讓每個函式變短。
+ 
+### 4. 命名與抽象層級
+ 
+- 禁止使用空泛命名:`data`／`process`／`manager`／`util`,
+  需具體對應業務語意(例如 `fetch_price_from_serp_api`,而非 `process_data`)。
+  `handle` 僅允許用於最外層協調邏輯(見第 5 節動詞詞彙表),其餘情況視為空泛命名
+- 不要引入未被使用的抽象層(interface／abstract class／factory／strategy),
+  除非確實存在 ≥ 2 種具體實作
+- 避免多層 wrapper／callback 包裝簡單邏輯
+
+### 5. 命名長度控制
+ 
+- 函式名應盡量控制在 2–4 個單字(動詞 + 名詞,必要時加修飾詞)
+- 不要在函式名重複「模組名／類別名／檔案名」已表達的語意
+  (例如在 `groq_summarizer.py` 內,函式命名為 `summarize()` 而非 `summarize_with_groq()`)
+- 使用固定動詞詞彙表:`get`／`fetch`(讀取)、`list`(多筆)、`build`／`make`(組裝)、
+  `validate`(驗證)、`normalize`(格式化)、`handle`(僅限最外層協調邏輯)
+- 過多細節(如條件、延遲、重試策略)用參數表達,不要塞進函式名稱
+- 只允許業界通用縮寫(`api`, `url`, `id`, `config`),業務核心詞彙不縮寫
+
+### 6. 換行與行長度
+ 
+- 單行程式碼上限為 100–120 字元(依語言慣例調整,Python 可用 100)
+- 函式呼叫參數、import、字典／物件字面量等,只有在超過行長度上限時才拆成多行,
+  未超過時保持單行
+- 不要僅因為「參數較多」或「表達式較長」就主動換行,以實際字元數為唯一判斷依據
+- 若專案已有 formatter 設定(如 black／prettier 的 line-length),
+  以該設定為準,不要用更保守的預設值
+
+### 7. 完成後自我審查
+ 
+完成程式碼後,請自問:
+ 
+1. 是否有函式混雜多種職責?→ 拆分(對照第 2、3 節)
+2. 是否有只被呼叫一次、拆分後反而要跳來跳去才看得懂的函式?→ 合併回去
+3. 是否有命名空泛、語意抽象、無法一眼看出用途的變數／函式?→ 重新命名(對照第 4、5 節)

Dockerfile

@@ -0,0 +1,27 @@
+# 建立docker image(鏡像)
+
+# 選擇base image(需符合 pyproject.toml 的 requires-python)
+FROM python:3.14-slim
+
+# 安裝 uv
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+
+# HF Space 以 uid 1000 執行,工作目錄需屬於該使用者才能寫入模型與資料
+RUN useradd -m -u 1000 user && mkdir -p /home/user/app && chown user:user /home/user/app
+USER user
+WORKDIR /home/user/app
+
+ENV UV_COMPILE_BYTECODE=1 \
+    UV_LINK_MODE=copy \
+    UV_PYTHON_DOWNLOADS=never \
+    PATH="/home/user/app/.venv/bin:$PATH"
+
+# 安裝套件依賴(先只複製依賴檔,讓此層可被快取)
+COPY --chown=user pyproject.toml uv.lock ./
+RUN uv sync --frozen --no-dev --no-install-project
+
+# 搬入原始碼
+COPY --chown=user . .
+
+# 執行cmd
+CMD ["uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,8 @@
+---
+title: Scam Detection
+emoji: 🔍
+colorFrom: red
+colorTo: blue
+sdk: docker
+sdk_version: "2.0"
+---

backend/__init__.py

@@ -0,0 +1 @@
+"""Scam Detection Backend Module"""

backend/api/exception_handler.py

@@ -0,0 +1,42 @@
+import logging
+from typing import cast
+
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from starlette.types import ExceptionHandler
+
+from backend.api.response import ApiErrorResponse
+
+logger = logging.getLogger(__name__)
+
+def make_error_response(status_code: int, message: object | None = None) -> JSONResponse:
+    """建立統一的錯誤回傳格式"""
+    response_content = ApiErrorResponse(
+        success=False,
+        version="1.0",
+        error_code=status_code,
+        error_message=message
+    )
+    return JSONResponse(status_code=status_code, content=jsonable_encoder(response_content))
+
+async def handle_global_exception(req: Request, exc: Exception) -> JSONResponse:
+    """處理全域錯誤"""
+    logger.error(f"Unhandled Exception: {str(exc)}", exc_info=True)
+    return make_error_response(status_code=500, message="Internal Server Error")
+
+async def handle_http_exception(req: Request, exc: HTTPException) -> JSONResponse:
+    """處理 HTTP 錯誤"""
+    logger.error(f"HTTP Exception: {str(exc.detail)}")
+    return make_error_response(status_code=exc.status_code, message=exc.detail)
+
+async def handle_validation_exception(req: Request, exc: RequestValidationError) -> JSONResponse:
+    """處理資料驗證錯誤"""
+    logger.error(f"Validation Error: {exc.errors()}")
+    return make_error_response(status_code=422, message=exc.errors())
+
+def register_exception_handlers(app: FastAPI) -> None:
+    app.add_exception_handler(Exception, handle_global_exception)
+    app.add_exception_handler(HTTPException, cast(ExceptionHandler, handle_http_exception))
+    app.add_exception_handler(RequestValidationError, cast(ExceptionHandler, handle_validation_exception))

backend/api/response.py

@@ -0,0 +1,27 @@
+"""建立後端統一回傳的格式"""
+from typing import Generic, TypeVar
+
+from pydantic import BaseModel, ConfigDict
+from pydantic.alias_generators import to_camel
+
+ResponseData = TypeVar("ResponseData")
+
+class ApiResponse(BaseModel, Generic[ResponseData]):
+    """統一回傳格式"""
+    success: bool = True
+    version: str = "1.0"
+    data: ResponseData | None = None
+
+class ApiErrorResponse(BaseModel):
+    model_config = ConfigDict(
+            alias_generator=to_camel,
+            populate_by_name=True,
+            serialize_by_alias=True,
+    )
+       
+    """統一回傳錯誤格式"""
+    success: bool = False
+    version: str = "1.0"
+    error_code: int
+    error_message: object | None = None
+    

backend/api/routers/image_price_validation.py

@@ -0,0 +1,130 @@
+"""商品圖片上傳、價格風險分析與 Google OCR 驗證 API。
+
+正式價格端點只負責 HTTP 輸入驗證、呼叫 image_price_service，並將完整 Service
+結果封裝成 ImagePriceResponse；OCR、商品辨識、價格統計與風險決策不在本模組。
+"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, File, HTTPException, Query, UploadFile
+from starlette.concurrency import run_in_threadpool
+
+from backend.api.response import ApiResponse
+from backend.api.schemas.image import (ImagePriceDebugInfo, ImagePriceResponse,
+                                       ImageUploadResponse,
+                                       MarketPriceEstimateResponse)
+from backend.core.config import settings
+from backend.services.image_price_service.image_price_analyzer import \
+    image_price_analyzer
+
+router = APIRouter(prefix="/v1", tags=["image-inference"])
+
+
+@router.post(
+    "/image/upload",
+    response_model=ApiResponse[ImageUploadResponse],
+    summary="圖片上傳測試",
+    description="用於測試圖片上傳是否成功。",
+)
+async def upload_image(file: UploadFile = File(...)) -> ApiResponse[ImageUploadResponse]:
+    """驗證圖片 MIME type、非空內容與大小，回傳上傳資訊但不執行分析。"""
+    content_type = file.content_type or ""
+    if content_type not in settings.ALLOWED_IMAGE_TYPES:
+        raise HTTPException(
+            status_code=400,
+            detail="僅支援 JPG、PNG、WEBP 圖片格式",
+        )
+
+    data = await file.read()
+    if not data:
+        raise HTTPException(status_code=400, detail="上傳檔案不可為空")
+
+    if len(data) > settings.MAX_IMAGE_BYTES:
+        raise HTTPException(status_code=400, detail="圖片大小不可超過 20MB")
+
+    data = ImageUploadResponse(
+        filename=file.filename or "unknown",
+        content_type=content_type,
+        size_bytes=len(data),
+        message="圖片已成功上傳，可進入下一步辨識流程",
+    )
+
+    return ApiResponse(data=data)
+
+
+@router.post(
+    "/price/analyze",
+    response_model=ApiResponse[ImagePriceResponse],
+    summary="價格驗證風險分析",
+    description="使用 OCR 分析圖片商品價格。加上 `?debug=true` 可取得 OCR 與市場價格來源的除錯資訊。",
+)
+async def analyze_image_price(
+    file: UploadFile = File(...),
+    include_debug: Annotated[bool | None, Query(alias="debug")] = None,
+) -> ApiResponse[ImagePriceResponse]:
+    """驗證商品圖片 HTTP 輸入並封裝 Service 分析結果。
+
+    Router 僅驗證 MIME type、非空內容與大小，呼叫 image_price_detector 後以
+    DTO 欄位建立 ImagePriceResponse，不計算或覆寫任何風險結果。
+    """
+    try:
+        content_type = file.content_type or ""
+        if content_type not in settings.ALLOWED_IMAGE_TYPES:
+            raise HTTPException(status_code=400, detail="僅支援 JPG、PNG、WEBP 圖片格式")
+
+        data = await file.read()
+        if not data:
+            raise HTTPException(status_code=400, detail="上傳檔案不可為空")
+        if len(data) > settings.MAX_IMAGE_BYTES:
+            raise HTTPException(status_code=400, detail="圖片大小不可超過 20MB")
+
+        result = await run_in_threadpool(
+            image_price_analyzer.image_price_detector,
+            data=data,
+            filename=file.filename or "unknown",
+            content_type=content_type,
+        )
+        warnings = list(result.extraction_warnings)
+        if not result.success and result.message and result.message not in warnings:
+            warnings.append(result.message)
+
+        debug_info = None
+        if include_debug:
+            debug_info = ImagePriceDebugInfo(
+                error_code=result.error_code,
+                search_tools=result.search_tools,
+                market_price_source=result.market_price_source,
+                market_price_estimates=tuple(
+                    MarketPriceEstimateResponse(
+                        **estimate.model_dump(exclude={"candidates", "search_tools"})
+                    )
+                    for estimate in result.market_price_estimates
+                ),
+                condition_source_text=result.condition_source_text,
+                condition_extraction_confidence=result.condition_extraction_confidence,
+                price_source_text=result.price_source_text,
+                extraction_confidence=result.extraction_confidence if result.success else None,
+                decision_layer=result.decision_layer,
+                warnings=warnings,
+            )
+
+        data=ImagePriceResponse(
+            product_name=result.product_name if result.success else None,
+            listed_price=result.listed_price if result.success else None,
+            market_price=(result.market_price if result.success and result.market_price > 0 else None),
+            seller_name=result.seller_name if result.success else None,
+            risk_label=result.risk_label,
+            condition=result.condition,
+            risk_score=result.score,
+            result=result.reason,
+            debug=debug_info,
+        )
+        
+        return ApiResponse(data=data)
+    
+    except HTTPException:
+        raise
+    except Exception as e:
+        if settings.DEBUG:
+            raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail="Internal Error")

backend/api/routers/phone_detection.py

@@ -0,0 +1,95 @@
+from fastapi import APIRouter, HTTPException
+
+from backend.api.response import ApiResponse
+from backend.api.schemas.phone import (PhoneQueryRequest, PhoneQueryResponse,
+                                       PhoneReportError, PhoneReportRequest,
+                                       PhoneReportResponse)
+from backend.services.phone_service import phone_service
+
+# 服務層為同步 MySQL I/O，路由用 def 讓 FastAPI 放進 threadpool 執行，避免阻塞 event loop
+router = APIRouter(prefix="/v1", tags=["phone"])
+
+
+@router.post(
+    "/phones/search",
+    response_model=ApiResponse[PhoneQueryResponse],
+    summary="電話號碼查詢",
+    description="查詢電話號碼是否已存在於黑白名單。若結果為 unknown，可提示使用者進行回報。",
+    openapi_extra={
+        "requestBody": {
+            "content": {
+                "application/json": {
+                    "example": {
+                        "phone_number": "0422255141"
+                    }
+                }
+            }
+        }
+    },
+)
+def query_phone(req: PhoneQueryRequest) -> ApiResponse[PhoneQueryResponse]:
+    try:
+        result = phone_service.query_phone(req.phone_number)
+        data = PhoneQueryResponse(**result)
+        return ApiResponse(data=data)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.post(
+    "/phones/report",
+    response_model=ApiResponse[PhoneReportResponse],
+    summary="可疑號碼回報",
+    description=(
+        "當查詢結果為 unknown 時，可回報該號碼為可疑名單。"
+        " 若選擇「其他」，請一併填寫 other_type 欄位。"
+    ),
+    openapi_extra={
+        "requestBody": {
+            "content": {
+                "application/json": {
+                    "examples": {
+                        "standard": {
+                            "summary": "回報常見類別",
+                            "value": {
+                                "phoneNumber": "0987654321",
+                                "phoneType": "假投資",
+                                "reporterPhone": "0912345678",
+                                "transferType": 1,
+                                "transferContent": "0912345678"
+                            }
+                        },
+                        "other": {
+                            "summary": "回報其他類別",
+                            "value": {
+                                "phoneNumber": "0987654321",
+                                "phoneType": "其他",
+                                "otherType": "疑似詐騙廣告",
+                                "reporterPhone": "0912345678",
+                                "transferType": 3,
+                                "transferContent": "https://example.com/fake"
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    },
+)
+def report_phone(req: PhoneReportRequest) -> ApiResponse[PhoneReportResponse]:
+    try:
+        result = phone_service.report_suspicious(
+            req.phone_number,
+            req.phone_type,
+            req.other_type,
+            None,
+            req.reporter_phone,
+            req.transfer_type,
+            req.transfer_content,
+        )
+        data = PhoneReportResponse(**result)
+        return ApiResponse(data=data)
+    except PhoneReportError as e:
+        raise HTTPException(status_code=409, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

backend/api/routers/text_inference.py

@@ -0,0 +1,46 @@
+from fastapi import APIRouter, HTTPException
+
+from backend.api.response import ApiResponse
+from backend.api.schemas.text import TextRequest, TextResponse
+from backend.core.config import settings
+from backend.services.text_service.text_analyzer import text_analyzer
+from backend.utils.text_cleaner import normalize_text
+
+router = APIRouter(prefix="/v1", tags=["text-inference"])
+
+
+@router.post("/text/analyze", response_model=ApiResponse[TextResponse], summary="文字詐騙風險分析", description="使用規則、多模型做混合判斷")
+async def analyze_text(req: TextRequest) -> ApiResponse[TextResponse]:
+    try:
+        text = normalize_text(req.text)
+        if not text:
+            raise HTTPException(status_code=400, detail="text 不可為空")
+
+        result = await text_analyzer.hybrid_detector(text)
+        data = TextResponse(**result)
+        return ApiResponse(data=data)
+    except HTTPException:
+        raise
+    except Exception as e:
+        if settings.DEBUG:
+            raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail="Internal Error")
+
+
+
+@router.post("/text/analyze/base", response_model=ApiResponse[TextResponse], summary="純模型文字風險分析", description="僅使用文字分類模型")
+def model_analyze_text(req: TextRequest) -> ApiResponse[TextResponse]:
+    try:
+        text = normalize_text(req.text)
+        if not text:
+            raise HTTPException(status_code=400, detail="text 不可為空")
+
+        result = text_analyzer.model_detector(text)
+        data = TextResponse(**result)
+        return ApiResponse(data=data)
+    except HTTPException:
+        raise
+    except Exception as e:
+        if settings.DEBUG:
+            raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail="Internal Error")

backend/api/routers/url_detection.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter, HTTPException
+
+from backend.api.response import ApiResponse
+from backend.api.schemas.url import UrlRequest, UrlResponse
+from backend.core.config import settings
+from backend.services.url_service.url_analyzer import url_detector
+
+router = APIRouter(prefix="/v1", tags=["url-detector"])
+
+# url網址分析
+@router.post("/url/analyze", response_model=ApiResponse[UrlResponse], summary="執行url分析")
+def analyze_url(req: UrlRequest) -> ApiResponse[UrlResponse]:
+    try:
+        url = str(req.url)
+        result = url_detector.analyze(url)
+        data = UrlResponse(**result)
+        return ApiResponse(data=data)
+    except Exception as e:
+        if settings.DEBUG:
+            raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail="Internal Error")
+    

backend/api/schemas/image.py

@@ -0,0 +1,71 @@
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_camel
+
+from backend.services.dto.price_analysis import (
+    DecisionLayer,
+    MarketPriceReferenceMode,
+    MarketPriceSource,
+    MarketPriceStatus,
+    SearchTool,
+)
+from backend.services.image_price_service.domain.models import MarketplaceCondition
+
+
+class ImageUploadResponse(BaseModel):
+    filename: str = Field(..., description="上傳檔名")
+    content_type: str = Field(..., description="檔案 MIME 類型")
+    size_bytes: int = Field(..., ge=1, description="檔案大小（byte）")
+    message: str = Field(..., description="處理結果訊息")
+
+
+class MarketPriceEstimateResponse(BaseModel):
+    """對外回傳且不含市場候選明細的價格區間。"""
+
+    model_config = ConfigDict(extra="forbid")
+
+    status: MarketPriceStatus
+    condition: MarketplaceCondition
+    reference_mode: MarketPriceReferenceMode
+    median_price: int = Field(ge=0)
+    low_price: int = Field(ge=0)
+    high_price: int = Field(ge=0)
+    sample_count: int = Field(ge=0)
+    site_count: int = Field(ge=0)
+    source: MarketPriceSource
+    confidence: float = Field(ge=0, le=1)
+
+
+class ImagePriceDebugInfo(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    error_code: str | None = None
+    search_tools: list[SearchTool] = Field(default_factory=list)
+    market_price_source: MarketPriceSource = "not_evaluated"
+    market_price_estimates: tuple[MarketPriceEstimateResponse, ...] = ()
+    condition_source_text: str = ""
+    condition_extraction_confidence: float = Field(default=0.0, ge=0, le=1)
+    price_source_text: str | None = None
+    extraction_confidence: float | None = Field(default=None, ge=0, le=1)
+    decision_layer: DecisionLayer
+    warnings: list[str] = Field(default_factory=list)
+
+
+class ImagePriceResponse(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        alias_generator=to_camel,
+        populate_by_name=True,
+        serialize_by_alias=True,
+    )
+
+    product_name: str | None = None
+    condition: MarketplaceCondition | None = None
+    listed_price: int | None = Field(default=None, ge=0)
+    market_price: int | None = Field(default=None, ge=0)
+    seller_name: str | None = None
+    risk_label: Literal["LOW", "MEDIUM", "HIGH", "UNKNOWN"]
+    risk_score: str | float | None = None
+    result: str | None = None
+    debug: ImagePriceDebugInfo | None = None

backend/api/schemas/phone.py

@@ -0,0 +1,124 @@
+import re
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic.alias_generators import to_camel
+
+
+class PhoneQueryRequest(BaseModel):
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+        serialize_by_alias=True,
+    )
+    
+    phone_number: str = Field(..., min_length=1, max_length=20)
+
+    @model_validator(mode="after")
+    def validate_phone_number_format(self):
+        if not re.fullmatch(r"[0-9]{8,15}", self.phone_number):
+            raise ValueError("電話號碼格式錯誤，請輸入 8~15 碼數字")
+        return self
+
+
+VALID_PHONE_TYPES = [
+    "約會交友",
+    "假投資",
+    "假信貸",
+    "假冒公務",
+    "假包裹釣魚",
+    "假求職",
+    "商業騷擾",
+    "假冒電商",
+    "其他",
+]
+
+
+class PhoneReportRequest(BaseModel):
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+        serialize_by_alias=True,
+    )
+
+    phone_number: str = Field(..., min_length=1, max_length=20)
+    phone_type: str = Field(..., min_length=1, max_length=50)
+    other_type: str | None = Field(None, max_length=100)
+    reporter_phone: str | None = Field(None, max_length=20)
+    transfer_type: int | None = Field(default=0)
+    transfer_content: str | None = Field(None, max_length=500)
+
+    @model_validator(mode="after")
+    def validate_phone_number_format(self):
+        if not re.fullmatch(r"[0-9]{8,15}", self.phone_number):
+            raise ValueError("電話號碼格式錯誤，請輸入 8~15 碼數字")
+        if self.reporter_phone and not re.fullmatch(r"[0-9]{8,15}", self.reporter_phone):
+            raise ValueError("回報者電話格式錯誤，請輸入 8~15 碼數字")
+        if self.transfer_type == 1 and self.transfer_content and not re.fullmatch(r"[0-9]{8,15}", self.transfer_content):
+            raise ValueError("轉介類型為電話號碼時，轉介內容必須為 8~15 碼數字")
+        return self
+
+    @model_validator(mode="after")
+    def validate_phone_type(self):
+        if self.phone_type not in VALID_PHONE_TYPES:
+            raise ValueError(
+                "phone_type 必須為其中之一：約會交友、假投資、假信貸、假冒公務、假包裹釣魚、假求職、商業騷擾、假冒電商、其他"
+            )
+        if self.phone_type == "其他" and not self.other_type:
+            raise ValueError("phone_type 為 '其他' 時，必須提供 other_type")
+        if self.transfer_type not in (0, 1, 2, 3, 4, 9):
+            raise ValueError("transfer_type 只能為 0、1、2、3、4、9，分別代表 無 / 電話 / LINE / URL / TG / 其他")
+        if self.transfer_type == 0 and self.transfer_content:
+            self.transfer_content = None
+        return self
+
+
+class FamilyStaticItems(BaseModel):
+    """電話家族的直接關聯證據"""
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+        serialize_by_alias=True,
+    )
+
+    related_phone: str = Field(..., description="關聯電話號碼")
+    weight: int = Field(..., ge=0, description="關聯強度分數")
+    reason: str = Field(..., description="關聯原因")
+    target_phone_type: str = Field(..., description="查詢號碼的詐騙類型")
+
+
+class PhoneQueryResponse(BaseModel):
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+        serialize_by_alias=True,
+    )
+
+    phone_number: str
+    status: str | None = None
+    phone_type: str | None = None
+    total_reports: int | None = None
+    first_reported_at: str | None = None
+    last_reported_at: str | None = None
+    owner_name: str | None = None
+    can_report: bool = False
+    # family analysis results (minimal but useful evidence)
+    family_static: list[FamilyStaticItems] | None = Field(
+        None,
+        description="直接關聯證據列表；如果沒有關聯，則為空陣列。",
+    )
+
+class PhoneReportResponse(BaseModel):
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+        serialize_by_alias=True,
+    )
+    phone_number: str
+    status: str
+    total_reports: int
+    report_time: str
+    message: str
+    connection_reason: str | None = None
+
+class PhoneReportError(Exception):
+    pass

backend/api/schemas/text.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_camel
+
+
+class TextRequest(BaseModel):
+    text: str = Field(..., min_length=1, max_length=10000)
+
+
+class TextResponse(BaseModel):
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+        serialize_by_alias=True,
+    )   
+    label: str
+    score: str | float | None = None
+    reason: str | None = None
+    cls_model_confidence: float | None = None

backend/api/schemas/url.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel, HttpUrl
+
+
+class UrlRequest(BaseModel):
+    url: HttpUrl
+
+class UrlResponse(BaseModel):
+    label: str
+    score: str | float | None = None
+    reason: str 

backend/core/config.py

@@ -0,0 +1,121 @@
+import os
+from pathlib import Path
+
+from pydantic import SecretStr, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+CORE_DIR = Path(__file__).resolve().parent
+BACKEND_DIR = CORE_DIR.parent
+ROOT_DIR = BACKEND_DIR.parent
+
+
+class Settings(BaseSettings):
+    APP_NAME: str = "Scam detetcor"
+    DEBUG: bool = True
+
+    HF_TEXT_REPO_ID: str = "scam-project/spam-detector-chinese"
+    HF_URL_REPO_ID: str = "scam-project/url-detector"
+    HF_PERSIST_REPO_ID: str = "scam-project/scam-rag-db"
+    HF_TOKEN: str = ""
+
+    REGEX_WEIGHT: float = 0.55
+    MODEL_WEIGHT: float = 0.45
+
+    HIGH_THRESHOLD: float = 0.5
+    MEDIUM_THRESHOLD: float = 0.6
+    UNKNOWN_THRESHOLD: float = 0.7
+
+    URL_HIGH_THRESHOLD: float = 0.5
+    URL_MEDIUM_THRESHOLD: float = 0.3
+    URL_MODEL_NAME: str = "url_scam_classifier.joblib"
+    URL_TW_RISK_BONUS: float = 0.3
+    URL_EXPAND_SHORTENED: bool = True
+    URL_EXPAND_TIMEOUT_SECONDS: float = 3.0
+    URL_EXPAND_MAX_REDIRECTS: int = 5
+
+    DEVICE: str = "cpu"
+    BASE_MODEL_PATH: str = str(BACKEND_DIR / "models")
+
+    CHUNCK_SIZE: int = 500
+    CHUNCK_OVERLAP: int = 50
+
+    RAG_MODEL_NAME: str = "openai/gpt-oss-20b"
+    RAG_ENABLED: bool = True
+    RAG_TOP_K: int = 3
+    RAG_RECORD_LIMIT: int = 2100
+    RAG_DATASET_PATH: str = str(ROOT_DIR / "data" / "raw" / "scam-dataset.json")
+    RAG_PERSIST_DIR: str = str(ROOT_DIR / "data" / "chroma")
+    EMBED_MODEL: str = "BAAI/bge-small-zh-v1.5"
+    
+    NORMALIZER_MODEL: str = "openai/gpt-oss-120b"
+    PRICE_MODEL: str = "openai/gpt-oss-120b"
+    REVIEW_MODEL: str = "qwen/qwen3.8-27b"
+    # 空字串代表不指定推理強度；gpt-oss 系列可用 low／medium／high
+    NORMALIZER_REASONING_EFFORT: str = "low"
+    PRICE_REASONING_EFFORT: str = "low"
+    PRICE_LLM_TIMEOUT_SECONDS: float = 10.0
+    PRICE_LLM_MAX_RETRIES: int = 1
+    TAVILY_SEARCH_API_KEY: SecretStr = SecretStr("")
+    SERP_API_KEY: SecretStr = SecretStr("")
+    SEARCH_COUNTRY: str = "taiwan"
+
+    ALLOWED_IMAGE_TYPES : set[str] = {"image/jpeg", "image/png", "image/webp"}
+    MAX_IMAGE_BYTES : int = 20 * 1024 * 1024  # 20MB
+    MARKET_PRICE_DB_PATH: str = str(ROOT_DIR / "data" / "market_prices_tw.json")
+    ONLINE_PRICE_ENABLED: bool = True
+    ONLINE_PRICE_MAX_RESULTS: int = 20
+    SEARCH_DOMAIN: list[str] = ["biggo.com.tw", "momo.com.tw", "pchome.com.tw", "shopee.tw", "ruten.com.tw", "tw.carousell.com"]
+    EXCLUDE_DOMAIN: list[str] = ["zh.wikipedia.org"]
+    SEARCH_TIMEOUT_SECONDS: float = 10.0
+    GROQ_FALLBACK_RETRY_DELAY_SECONDS: float = 2.0
+    GROQ_RATE_LIMIT_MAX_WAIT_SECONDS: float = 30.0
+    PRODUCT_MATCH_FUZZY_MIN_SCORE: int = 86
+    ENABLE_INTELLIGENT_LAYER: bool = True
+    CASE_LOG_PATH: str = str(ROOT_DIR / "data" / "cases" / "image_analysis_cases.jsonl")
+    CASE_MEMORY_ENABLED: bool = True
+    OCR_PROVIDER: str = "google_vision"
+    GCV_LANGUAGE_HINTS: str = "zh-TW,en"
+    GCP_OCR_SERVICE_ACCOUNT_JSON : str = ""
+
+    DB_USERNAME: str = ""
+    DB_PASSWORD: str = ""
+    DB_HOST: str = ""
+    DB_PORT: int | None = None
+    DB_NAME: str = ""
+    DB_TIMEOUT: int = 10
+    DB_SSL_CA: str = ""
+
+    GROQ_API_KEY: SecretStr = SecretStr("")
+    model_config = SettingsConfigDict(
+        env_file=[os.path.join(BACKEND_DIR, ".env"), 
+                  os.path.join(ROOT_DIR, ".env")],
+        extra="ignore",
+    )
+    
+    @field_validator("DEBUG", mode="before")
+    @classmethod
+    def parse_debug(cls, value: object) -> object:
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in {"release", "prod", "production"}:
+                return False
+            if normalized in {"debug", "dev", "development"}:
+                return True
+        return value
+
+    @field_validator("DB_PORT", mode="before")
+    @classmethod
+    def parse_db_port(cls, value: object) -> object:
+        if isinstance(value, str) and not value.strip():
+            return None
+        return value
+
+    @field_validator("DB_SSL_CA")
+    @classmethod
+    def resolve_db_ssl_ca(cls, value: str) -> str:
+        if value and not os.path.isabs(value):
+            return str(ROOT_DIR / value)
+        return value
+
+
+settings = Settings()

backend/core/lifespan.py

@@ -0,0 +1,51 @@
+import logging
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from langchain_chroma import Chroma
+from langchain_huggingface import HuggingFaceEmbeddings
+
+from backend.core.config import settings
+from backend.rag.rag_context import RAGContext
+from backend.rag.rag_reasoner import get_rag_chain
+from backend.rag.rag_retriever import is_rag_ready
+from backend.services.text_service.base_classifier import base_classifier
+from backend.services.url_service.url_analyzer import url_detector
+
+logger = logging.getLogger(__name__)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    print("應用環境啟動")
+
+    base_classifier.load_model()
+    url_detector.load_model()
+
+    # 預熱：ONNX session 首次推論與 tldextract 後綴表載入（可能需連網）都很慢，
+    # 在啟動時先跑一次，避免由第一個使用者請求承擔
+    base_classifier.predict_text("預熱")
+    url_detector.analyze("https://www.example.com/")
+
+    print("初始化 RAG")
+    RAGContext.set_app(app)
+    if not is_rag_ready():
+        RAGContext.load_vectorstore()
+
+    embeddings = HuggingFaceEmbeddings(model_name=settings.EMBED_MODEL)
+    embeddings.embed_query("warmup")
+    app.state.embeddings = embeddings
+    app.state.vectorstore = Chroma(
+        persist_directory=settings.RAG_PERSIST_DIR, embedding_function=embeddings
+    )
+
+    if is_rag_ready():
+        try:
+            get_rag_chain()
+        except Exception:
+            # 例如尚未設定 GROQ_API_KEY；保留延遲建立，第一次 RAG 請求時再嘗試
+            logger.warning("RAG chain 預先建立失敗，改為首次請求時建立", exc_info=True)
+    print("RAG 已初始化")
+
+    yield
+    print("應用關閉")

backend/core/logging.py

@@ -0,0 +1,10 @@
+import logging
+
+
+def configure_logging():
+    """配置日誌系統"""
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(levelname)s:%(name)s:%(message)s",
+        force=True
+)

backend/main.py

@@ -0,0 +1,40 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from backend.api.exception_handler import register_exception_handlers
+from backend.api.routers import (image_price_validation, phone_detection,
+                                 text_inference, url_detection)
+from backend.core.lifespan import lifespan
+from backend.core.logging import configure_logging
+
+configure_logging()
+
+# 主入口
+app = FastAPI(
+    lifespan=lifespan,
+    title="詐騙風險偵測API",
+    description="用於檢測詐騙風險的API",
+    version="1.0.0"    
+)
+
+# 註冊全域例外處理
+register_exception_handlers(app)
+
+# CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 掛載router
+app.include_router(text_inference.router, prefix="/api")
+app.include_router(url_detection.router, prefix="/api")
+app.include_router(image_price_validation.router, prefix="/api")
+app.include_router(phone_detection.router, prefix="/api")
+
+@app.get("/")
+async def root():
+    return {"message": "AI API is running..."}

backend/persistence/mysql_connection.py

@@ -0,0 +1,52 @@
+from typing import Any
+
+from sqlalchemy import create_engine
+from sqlalchemy.engine import URL
+from sqlalchemy.orm import sessionmaker
+
+from backend.core.config import settings
+
+host_name = settings.DB_HOST.strip()
+port_number = settings.DB_PORT
+if host_name and ":" in host_name and host_name.count(":") == 1:
+    host_candidate, port_candidate = host_name.rsplit(":", 1)
+    if port_candidate.isdigit():
+        host_name = host_candidate
+        port_number = int(port_candidate)
+
+database_url = URL.create(
+    drivername="mysql+pymysql",
+    username=settings.DB_USERNAME,
+    password=settings.DB_PASSWORD,
+    host=host_name,
+    port=port_number,
+    database=settings.DB_NAME,
+    query={"charset": "utf8mb4"},
+)
+
+ssl_config = {"ca": settings.DB_SSL_CA} if settings.DB_SSL_CA else {}
+
+connect_args: dict[str, Any] = {"connect_timeout": settings.DB_TIMEOUT}
+if ssl_config:
+    connect_args["ssl"] = ssl_config
+
+engine = create_engine(
+    database_url,
+    pool_pre_ping=True,
+    pool_recycle=3600,
+    connect_args=connect_args,
+)
+
+SessionLocal = sessionmaker(
+    autocommit=False,
+    autoflush=False,
+    bind=engine,
+)
+
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

backend/providers/__init__.py

@@ -0,0 +1,5 @@
+"""外部模型與服務供應者。"""
+
+from backend.providers.groq import GroqProvider, groq_provider
+
+__all__ = ["GroqProvider", "groq_provider"]

backend/providers/groq.py

@@ -0,0 +1,83 @@
+"""集中建立 Groq 聊天模型。"""
+
+import os
+from typing import Any
+
+from langchain_core.language_models.chat_models import BaseChatModel
+from pydantic import SecretStr
+
+from backend.core.config import settings
+
+
+class GroqProvider:
+    """建立具有一致 API key 處理方式的 Groq 聊天模型。"""
+
+    @staticmethod
+    def create(
+        *,
+        model: str,
+        temperature: float = 0,
+        api_key: str | SecretStr | None = None,
+        reasoning_effort: str | None = None,
+        timeout: float | None = None,
+        max_retries: int | None = None,
+        model_kwargs: dict[str, Any] | None = None,
+    ) -> BaseChatModel:
+        """依用途建立 Groq 模型，僅傳入有設定的選用參數。"""
+        from langchain_groq import ChatGroq
+
+        kwargs: dict[str, Any] = {
+            "model": model,
+            "temperature": temperature,
+        }
+        secret = GroqProvider.resolve_api_key(api_key)
+        if secret is not None:
+            kwargs["api_key"] = secret
+        if reasoning_effort:
+            kwargs["reasoning_effort"] = reasoning_effort
+        if timeout is not None:
+            kwargs["timeout"] = timeout
+        if max_retries is not None:
+            kwargs["max_retries"] = max_retries
+        if model_kwargs:
+            kwargs["model_kwargs"] = dict(model_kwargs)
+
+        return ChatGroq(**kwargs)
+
+    @classmethod
+    def resolve_api_key(
+        cls,
+        value: str | SecretStr | None = None,
+    ) -> SecretStr | None:
+        """依序解析指定值、應用設定與環境變數中的 Groq API key。"""
+        secret = cls._secret(value)
+        if secret is not None:
+            return secret
+
+        secret = cls._secret(settings.GROQ_API_KEY)
+        if secret is not None:
+            return secret
+
+        return cls._secret(os.getenv("GROQ_API_KEY"))
+
+    @classmethod
+    def is_configured(
+        cls,
+        value: str | SecretStr | None = None,
+    ) -> bool:
+        """確認是否有可用的 Groq API key。"""
+        return cls.resolve_api_key(value) is not None
+
+    @staticmethod
+    def _secret(value: str | SecretStr | None) -> SecretStr | None:
+        """將單一非空 API key 正規化為 SecretStr。"""
+        if isinstance(value, SecretStr):
+            return value if value.get_secret_value() else None
+        if isinstance(value, str) and value.strip():
+            return SecretStr(value.strip())
+        return None
+
+
+groq_provider = GroqProvider()
+
+__all__ = ["GroqProvider", "groq_provider"]

backend/rag/__init__.py

@@ -0,0 +1 @@
+"""RAG (Retrieval-Augmented Generation) module"""

backend/rag/dto/rag_analysis.py

@@ -0,0 +1,21 @@
+from pydantic import BaseModel, Field
+
+
+class RagEvidence(BaseModel):
+    used: bool
+    label: str | None = None
+    score: float | None = None
+    reason: str | None = None
+    raw_response: str | None = None
+
+
+class RagResponse(BaseModel):
+    urgency: float = Field(ge=0, le=1, description="催促語氣")
+    money_related: float = Field(ge=0, le=1, description="金錢獲利")
+    baiting: float = Field(ge=0, le=1, description="異常誘導")
+    asks_for_personal_info: float = Field(ge=0, le=1, description="索取個資")
+    reputation: float = Field(ge=0, le=1, description="不可信度")
+    is_anti_fraud: float = Field(ge=0, le=1, description="是否為反詐騙")
+    social_engineering: float = Field(default=0.0, ge=0, le=1, description="社交工程和情緒操控")
+    reason: str = Field(..., min_length=50, max_length=300, description="原因")
+    suggestion: str = Field(..., min_length=10, max_length=30, description="建議")

backend/rag/rag_context.py

@@ -0,0 +1,52 @@
+import os
+from typing import Any
+
+from huggingface_hub import snapshot_download
+
+from backend.core.config import settings
+
+
+class RAGContext:
+    """全局 RAG 上下文，存儲 app 實例"""
+    
+    _app: Any = None
+
+    @staticmethod
+    def load_vectorstore() -> None:
+        """從 HuggingFace 加載向量資料庫"""
+        print("加載向量資料庫...")
+        snapshot_download(
+            repo_id=settings.HF_PERSIST_REPO_ID,
+            repo_type="dataset",
+            local_dir=settings.RAG_PERSIST_DIR,
+            token=settings.HF_TOKEN or None
+        )
+        print("向量資料庫加載完畢")
+
+    @classmethod
+    def set_app(cls, app: Any) -> None:
+        """設置 FastAPI app 實例"""
+        cls._app = app
+    
+    @classmethod
+    def get_app(cls) -> Any:
+        """取得 FastAPI app 實例"""
+        if cls._app is None:
+            raise RuntimeError("RAG Context 未初始化，請先呼叫 set_app()")
+        return cls._app
+    
+    @classmethod
+    def get_vectorstore(cls) -> Any:
+        """取得 vectorstore"""
+        app = cls.get_app()
+        if not hasattr(app.state, 'vectorstore'):
+            raise RuntimeError("vectorstore 尚未初始化")
+        return app.state.vectorstore
+    
+    @classmethod
+    def get_embeddings(cls) -> Any:
+        """取得 embeddings"""
+        app = cls.get_app()
+        if not hasattr(app.state, 'embeddings'):
+            raise RuntimeError("embeddings 尚未初始化")
+        return app.state.embeddings

backend/rag/rag_reasoner.py

@@ -0,0 +1,178 @@
+import json
+from typing import Any
+
+from backend.core.config import settings
+from backend.providers import groq_provider
+from backend.rag.dto.rag_analysis import RagEvidence, RagResponse
+from backend.rag.rag_retriever import (format_context, get_retriever,
+                                       is_rag_ready, normalize_query_text)
+from backend.utils.text_cleaner import normalize_escape_sequences
+
+_rag_chain : Any | None = None
+
+def _require_langchain() -> tuple[Any, Any, Any]:
+    from langchain_core.output_parsers import JsonOutputParser
+    from langchain_core.prompts import ChatPromptTemplate
+    from langchain_core.runnables import RunnablePassthrough
+
+    return ChatPromptTemplate, RunnablePassthrough, JsonOutputParser
+
+
+def _get_prompt() -> Any:
+    ChatPromptTemplate, _, _ = _require_langchain()
+    template = """
+        你是一個防詐風險分析專家，請根據訊息內容分析不同風險特徵的符合程度。
+
+        【參考案例 - 僅供理解詐騙模式，禁止複製其中文字到 reason】
+        {context}
+        【參考案例結束】
+        
+        【待分析訊息 - 僅供理解內容，禁止複製其中文字到 reason】
+        {question}
+        【待分析訊息結束】
+
+        請依序完成以下判斷：
+        第一步：判斷這則訊息的主要目的
+        - A：提醒他人避免詐騙、警示風險、安全建議
+        - B：可疑訊息（詐騙、誘導、索資、施壓、領錢、假冒通知）
+        - C：一般安全訊息（正常日常對話、正常通知）
+        - D：對話類互動異常（對話中出現一方威脅、一方示弱、且具有情緒操控）
+        
+        第二步：
+        - 若第一步為 A，分析以下維度（0~1）：
+        - is_anti_fraud：是否具有反詐騙、教育勸導 
+        直接輸出下方 JSON，reason，其他欄位全部填 0.0，用繁體中文說明，禁止複製原文
+        {format_instructions}
+        
+        - 若第一步為 B 或 C，再分析以下維度（0~1）：
+        - urgency：催促、限時、要求立即操作
+        - money_related：誘導金錢交易、投資、匯款，通知領錢
+        - baiting：不合理優惠、保證獲利、隱藏交易
+        - asks_for_personal_info：要求個資、帳號、驗證碼
+        - reputation：來源可疑、假冒權威、誇大承諾
+        - reason：禁止複製原文，用繁體中文說明訊息中出現哪些疑點
+        - suggestion：針對此風險給出一句具體的下一步建議，以「建議」開頭
+
+        - 若第一步為 D，分析以下維度（0~1）：
+        - urgency：是否出現威脅、限時、製造恐慌
+        - money_related：是否涉及借錢、投資、轉帳請求，通知領錢
+        - baiting：是否以感情、同情、利益逐步誘導
+        - asks_for_personal_info：是否套問個資、帳號、行蹤
+        - reputation：身份是否可疑、前後矛盾、來路不明
+        - social_engineering：是否存在情緒操控、威脅（0~1）
+        - reason：禁止複製原文，用繁體中文具體說明對話中出現哪些可疑行為或話術
+        - suggestion：針對此風險給出一句具體的下一步建議，以「建議」開頭
+        輸出格式：
+        {format_instructions}
+    """
+
+    parser = _get_parser()
+    prompt = ChatPromptTemplate.from_template(
+        template, 
+        partial_variables={"format_instructions": parser.get_format_instructions()}
+    )
+    return prompt
+
+def _get_llm() -> Any:
+    return groq_provider.create(
+        model=settings.RAG_MODEL_NAME,
+        temperature=0.1,
+        model_kwargs={
+            "top_p": 0.7,
+        }
+    )
+
+
+def _get_parser() -> Any:
+    _, _, JsonOutputParser = _require_langchain()
+    parser = JsonOutputParser(pydantic_object=RagResponse)
+    return parser
+
+def get_rag_chain() -> Any:
+    global _rag_chain
+    if _rag_chain is None:
+        _rag_chain = build_rag_chain()
+    return _rag_chain
+
+def build_rag_chain() -> Any:
+    _, RunnablePassthrough, _ = _require_langchain()
+
+    prompt = _get_prompt()
+    retriever = get_retriever()
+    llm = _get_llm()
+    parser = _get_parser()
+
+    return (
+        {
+            "context": retriever | format_context,
+            "question": RunnablePassthrough() | normalize_query_text,
+        }
+        | prompt
+        | llm
+        | parser
+    )
+
+
+def _to_float(v: Any) -> float:
+    """安全轉換任意值為浮點數，預設為 0.0"""
+    try:
+        return float(v)
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def parse_structured_response(text: str) -> tuple[str, float, str]:
+    payload = json.loads(text)
+    if not isinstance(payload, dict):
+        return "回傳錯誤", -1, ""
+    money = _to_float(payload.get("money_related"))
+    urgency = _to_float(payload.get("urgency"))
+    baiting = _to_float(payload.get("baiting"))
+    personal_info = _to_float(payload.get("asks_for_personal_info"))
+    is_anti_fraud = _to_float(payload.get("is_anti_fraud"))
+    social_engineering = _to_float(payload.get("social_engineering"))
+    reputation = _to_float(payload.get("reputation"))
+    
+    print(money, urgency, baiting, personal_info, is_anti_fraud, social_engineering, reputation, end='\n')
+
+    score = min((money * 0.25 + 
+                urgency * 0.30 + 
+                baiting * 0.25 + 
+                personal_info * 0.25 +
+                social_engineering * 0.30 -
+                is_anti_fraud * 1 +
+                reputation * 0.30) * 100, 100)
+    
+    score = max(score, 5)
+    suggestion = normalize_escape_sequences(str(payload.get("suggestion") or ""))
+    reason = normalize_escape_sequences(str(payload.get("reason") or ""))
+    
+    label = "低風險"
+    if score >= 80:
+        label = "高風險"
+    elif score >= 40:
+        label = "中等風險"
+    
+    return label, score, f"{reason}{suggestion}"
+
+
+async def analyze_with_rag(message: str) -> RagEvidence:
+    if not is_rag_ready():
+        return RagEvidence(used=False)
+
+    rag_chain = get_rag_chain()
+    # 避免被同步請求卡住
+    chain_output = await rag_chain.ainvoke(message)
+    if isinstance(chain_output, dict):
+        raw_response = json.dumps(chain_output, ensure_ascii=False)
+    else:
+        raw_response = chain_output
+    label, score, reason = parse_structured_response(raw_response)
+
+    return RagEvidence(
+        used=True,
+        label=label,
+        score=score,
+        reason=reason,
+        raw_response=raw_response,
+    )

backend/rag/rag_retriever.py

@@ -0,0 +1,43 @@
+from pathlib import Path
+from typing import Any
+
+from backend.core.config import settings
+from backend.rag.rag_context import RAGContext
+from backend.utils.text_cleaner import normalize_text
+
+
+def _get_vectorstore() -> Any:
+    try:
+        return RAGContext.get_vectorstore()
+    except RuntimeError as e:
+        print(f"取得 vectorstore 失敗: {e}")
+        raise
+
+
+def get_retriever() -> Any:
+    vectorstore = _get_vectorstore()
+    return vectorstore.as_retriever(search_kwargs={"k": settings.RAG_TOP_K})
+
+
+def format_context(docs: list[Any]) -> str:
+    blocks: list[str] = []
+    for idx, doc in enumerate(docs, start=1):
+        blocks.append(f"[相似案例 {idx}]\n{doc.page_content[:190]}") # 限制每句參考在190以內
+    return "\n\n".join(blocks)
+
+
+def normalize_query_text(message: str) -> str:
+    cleaned = normalize_text(message)
+    return f"待分析訊息: {cleaned}"
+
+
+def retrieve_similar_cases(message: str) -> list[Any]:
+    if not is_rag_ready():
+        return []
+    return get_retriever().invoke(normalize_query_text(message))
+
+
+def is_rag_ready() -> bool:
+    if not settings.RAG_ENABLED:
+        return False
+    return Path(settings.RAG_PERSIST_DIR).exists()

backend/repository/case_repository.py

@@ -0,0 +1,28 @@
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import uuid4
+
+from backend.core.config import settings
+
+
+class CaseRepository:
+    def __init__(self, path: str | None = None):
+        self._path = Path(path or settings.CASE_LOG_PATH)
+
+    def save(self, payload: dict) -> str:
+        case_id = str(uuid4())
+        envelope = {
+            "case_id": case_id,
+            "created_at": datetime.now(timezone.utc).isoformat(),
+            "payload": payload,
+        }
+
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        with self._path.open("a", encoding="utf-8") as f:
+            f.write(json.dumps(envelope, ensure_ascii=False) + "\n")
+
+        return case_id
+
+
+case_repository = CaseRepository()

backend/repository/family_repository.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+
+from sqlalchemy import bindparam, text
+
+from backend.persistence.mysql_connection import engine as db_engine
+from backend.utils.phone_normalizer import normalize, normalize_db_phone
+
+_SELF_ROW_QUERY = text(
+    """
+    SELECT phone_number,
+           dominant_tag,
+           first_seen_at,
+           last_seen_at,
+           referral_type,
+           referral_info
+    FROM blacklist
+    WHERE phone_number IN (:phone_db, :phone_full)
+    LIMIT 1
+    """
+)
+
+_RELATED_ROWS_QUERY = text(
+    """
+    SELECT DISTINCT phone_number,
+           dominant_tag AS phone_type,
+           first_seen_at AS first_reported_at,
+           last_seen_at AS last_reported_at,
+           referral_type AS transfer_type,
+           referral_info AS transfer_content
+    FROM blacklist
+    WHERE phone_number NOT IN (:phone_db, :phone_full)
+      AND (
+            referral_info IN (:phone_db, :phone_full)
+            OR LEFT(phone_number, 6) = LEFT(:phone_db, 6)
+            OR dominant_tag = :phone_type
+          )
+    ORDER BY
+      CASE
+        WHEN referral_info IN (:phone_db, :phone_full) THEN 0
+        WHEN LEFT(phone_number, 6) = LEFT(:phone_db, 6) THEN 1
+        ELSE 2
+      END
+    LIMIT 20
+    """
+)
+
+_REPORTER_PHONES_QUERY = text(
+    """
+    SELECT DISTINCT reporter_phone
+    FROM fraud_reports
+    WHERE phone_number IN (:phone_db, :phone_full)
+    LIMIT 10
+    """
+)
+
+_REPORTER_ROWS_QUERY = text(
+    """
+    SELECT DISTINCT fr.phone_number,
+           fr.tag AS phone_type,
+           fr.report_time AS first_reported_at,
+           fr.report_time AS last_reported_at,
+           fr.referral_type AS transfer_type,
+           fr.referral_info AS transfer_content
+    FROM fraud_reports fr
+    WHERE fr.reporter_phone IN :reporter_phone_list
+    LIMIT 20
+    """
+).bindparams(bindparam("reporter_phone_list", expanding=True))
+
+# 只撈指定號碼的通報事件；原本每次查詢都整張表載入，資料量越大越慢
+_REPORT_EVENTS_QUERY = text(
+    """
+    SELECT reporter_phone, phone_number, report_time
+    FROM fraud_reports
+    WHERE phone_number IN :phone_list
+      AND reporter_phone IS NOT NULL
+      AND report_time IS NOT NULL
+    """
+).bindparams(bindparam("phone_list", expanding=True))
+
+
+class FamilyRepository:
+    def __init__(self, engine: Any | None = None):
+        self._engine = engine or db_engine
+
+    def list_fraud_report_events(
+        self, phones: Iterable[str], engine: Any | None = None
+    ) -> dict[str, list[tuple[str, Any]]]:
+        """回傳 {通報者: [(被通報號碼, 通報時間), ...]}，只包含 phones 內號碼的事件。
+
+        號碼同時以完整格式與資料庫格式（去前導 0）查詢，與本模組其他查詢的假設一致。
+        """
+        active_engine = engine or self._engine
+        phone_list = sorted(
+            {form for phone in phones for form in (phone, normalize_db_phone(phone)) if form}
+        )
+        if active_engine is None or not phone_list:
+            return {}
+
+        with active_engine.connect() as conn:
+            rows = conn.execute(_REPORT_EVENTS_QUERY, {"phone_list": phone_list}).mappings().all()
+
+        events: dict[str, list[tuple[str, Any]]] = {}
+        for row in rows:
+            reporter = normalize(row.get("reporter_phone"))
+            phone = normalize(row.get("phone_number"))
+            reported_at = row.get("report_time")
+            if reporter and phone and reported_at:
+                events.setdefault(reporter, []).append((phone, reported_at))
+        return events
+
+    def list_family_records(
+        self, phone_number: str, engine: Any | None = None
+    ) -> list[dict[str, Any]]:
+        """撈出查詢號碼本身、可能相關的黑名單號碼與同通報者通報過的號碼，作為家族分析輸入。"""
+        active_engine = engine or self._engine
+        normalized_phone = normalize(phone_number)
+        if active_engine is None or not normalized_phone:
+            return []
+
+        phone_params = {
+            "phone_db": normalize_db_phone(phone_number),
+            "phone_full": normalized_phone,
+        }
+        with active_engine.connect() as conn:
+            self_row = conn.execute(_SELF_ROW_QUERY, phone_params).mappings().first()
+            phone_type = self_row["dominant_tag"] if self_row else "其他"
+            related_rows = conn.execute(
+                _RELATED_ROWS_QUERY, {**phone_params, "phone_type": phone_type}
+            ).mappings().all()
+
+            reporter_phones = [
+                row["reporter_phone"]
+                for row in conn.execute(_REPORTER_PHONES_QUERY, phone_params).mappings().all()
+                if row.get("reporter_phone")
+            ]
+            reporter_rows = []
+            if reporter_phones:
+                reporter_rows = conn.execute(
+                    _REPORTER_ROWS_QUERY, {"reporter_phone_list": reporter_phones}
+                ).mappings().all()
+
+        records: list[dict[str, Any]] = []
+        if self_row:
+            records.append({
+                "電話號碼": self_row["phone_number"],
+                "通報標籤": self_row["dominant_tag"] or "其他",
+                "首次通報時間": self_row["first_seen_at"],
+                "最後通報時間": self_row["last_seen_at"],
+                "轉介類型": self_row["referral_type"] or 0,
+                "轉介內容": self_row["referral_info"] or "",
+                "通報者識別碼": "",
+                "通報次數": 1,
+            })
+        for entry in [*related_rows, *reporter_rows]:
+            records.append({
+                "電話號碼": entry["phone_number"],
+                "通報標籤": entry["phone_type"] or "其他",
+                "首次通報時間": entry["first_reported_at"],
+                "最後通報時間": entry["last_reported_at"],
+                "轉介類型": entry["transfer_type"] or 0,
+                "轉介內容": entry["transfer_content"] or "",
+                "通報者識別碼": "",
+                "通報次數": 1,
+            })
+        return records
+
+
+family_repository = FamilyRepository()

backend/repository/market_price_repository.py

@@ -0,0 +1,125 @@
+import json
+import re
+import unicodedata
+from pathlib import Path
+from typing import TypedDict
+
+from rapidfuzz import fuzz
+
+from backend.core.config import settings
+
+
+class MarketPriceRecord(TypedDict):
+    aliases: list[str]
+    product_name: str
+    brand_model: str
+    market_price: int
+
+
+DEFAULT_MARKET_PRICE_RECORDS: list[MarketPriceRecord] = [
+    {
+        "aliases": [
+            "Panasonic EH-NE11",
+            "國際牌 EH-NE11",
+            "EH-NE11",
+            "Panasonic 國際牌 1200W 負離子速乾型冷熱吹風機",
+        ],
+        "product_name": "Panasonic 國際牌 1200W 負離子速乾型冷熱吹風機",
+        "brand_model": "Panasonic EH-NE11",
+        "market_price": 1290,
+    },
+    {
+        "aliases": ["iPhone 15", "Apple iPhone 15"],
+        "product_name": "Apple iPhone 15",
+        "brand_model": "Apple iPhone 15",
+        "market_price": 27900,
+    },
+]
+
+
+class MarketPriceRepository:
+    def __init__(self, source_path: str | None = None):
+        self._source_path = Path(source_path or settings.MARKET_PRICE_DB_PATH)
+        self._records = self._load_records()
+        self._alias_index = self._build_alias_index(self._records)
+
+    @classmethod
+    def _build_alias_index(
+        cls, records: list[MarketPriceRecord]
+    ) -> list[tuple[MarketPriceRecord, str, str]]:
+        """預先算好每個別名的精確比對與模糊比對正規化結果，避免每次查詢重算。"""
+        return [
+            (item, cls._normalize(alias), cls._normalize_for_fuzzy(alias))
+            for item in records
+            for alias in item["aliases"]
+        ]
+
+    @staticmethod
+    def _normalize(text: str) -> str:
+        text = unicodedata.normalize("NFKC", text).lower()
+        return re.sub(r"[^a-z0-9\u4e00-\u9fff]+", "", text)
+
+    @staticmethod
+    def _normalize_for_fuzzy(text: str) -> str:
+        text = unicodedata.normalize("NFKC", text).lower()
+        text = re.sub(r"[^a-z0-9\u4e00-\u9fff]+", " ", text)
+        return re.sub(r"\s+", " ", text).strip()
+
+    def _load_records(self) -> list[MarketPriceRecord]:
+        if not self._source_path.exists():
+            return DEFAULT_MARKET_PRICE_RECORDS.copy()
+
+        with self._source_path.open("r", encoding="utf-8") as f:
+            raw = json.load(f)
+
+        records: list[MarketPriceRecord] = []
+        for row in raw:
+            aliases = [str(a) for a in row.get("aliases", []) if str(a).strip()]
+            if not aliases:
+                continue
+
+            market_price = int(row.get("market_price", 0) or 0)
+            records.append(
+                {
+                    "aliases": aliases,
+                    "product_name": str(row.get("product_name", "未知商品")),
+                    "brand_model": str(row.get("brand_model", "未知型號")),
+                    "market_price": market_price,
+                }
+            )
+
+        return records
+
+    def reload(self) -> None:
+        self._records = self._load_records()
+        self._alias_index = self._build_alias_index(self._records)
+
+    def find_by_text(self, text: str) -> tuple[str, str, int]:
+        normalized = self._normalize(text)
+        for item, alias_exact, _ in self._alias_index:
+            if alias_exact in normalized:
+                return item["product_name"], item["brand_model"], item["market_price"]
+
+        # OCR 容錯：若有輕微辨識錯字，使用模糊比對兜底。
+        fuzzy_text = self._normalize_for_fuzzy(text)
+        best_score = 0.0
+        best_item: MarketPriceRecord | None = None
+
+        for item, _, alias_fuzzy in self._alias_index:
+            if not alias_fuzzy:
+                continue
+
+            score = max(
+                float(fuzz.partial_ratio(alias_fuzzy, fuzzy_text)),
+                float(fuzz.token_set_ratio(alias_fuzzy, fuzzy_text)),
+            )
+            if score > best_score:
+                best_score = score
+                best_item = item
+
+        if best_item and best_score >= settings.PRODUCT_MATCH_FUZZY_MIN_SCORE:
+            return best_item["product_name"], best_item["brand_model"], best_item["market_price"]
+
+        return "未知商品", "未知型號", 0
+    
+market_price_repository = MarketPriceRepository()

backend/repository/phone_repository.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.exc import OperationalError
+
+from backend.persistence.mysql_connection import engine as db_engine
+from backend.utils.phone_normalizer import normalize, normalize_db_phone
+
+
+class PhoneRepository:
+    def __init__(self, engine: Any | None = None):
+        self._engine = engine or db_engine
+
+    def get_phone_record(self, phone_number: str) -> dict[str, Any] | None:
+        if self._engine is None:
+            return None
+
+        normalized_phone = normalize(phone_number)
+        db_phone = normalize_db_phone(phone_number)
+        query = text(
+            """
+            SELECT p.phone_number,
+                   p.tags,
+                   b.report_count AS total_reports,
+                   b.first_seen_at AS first_reported_at,
+                   b.last_seen_at AS last_reported_at,
+                   b.dominant_tag AS phone_type,
+                   w.org_name AS owner_name
+            FROM phone p
+            LEFT JOIN blacklist b ON p.phone_number = b.phone_number
+            LEFT JOIN whitelist w ON p.phone_number = w.phone_number
+            WHERE p.phone_number IN (:phone_db, :phone_full)
+            """
+        )
+
+        try:
+            with self._engine.begin() as conn:
+                return conn.execute(
+                    query,
+                    {"phone_db": db_phone, "phone_full": normalized_phone},
+                ).mappings().first()
+        except OperationalError:
+            return None
+
+    def report_suspicious(
+        self,
+        phone_number: str,
+        phone_type: str,
+        reporter_phone: str | None,
+        transfer_type: int | None = 0,
+        transfer_content: str | None = None,
+    ) -> dict[str, Any]:
+        if self._engine is None:
+            raise RuntimeError("資料庫連線失敗：database engine unavailable")
+
+        normalized_phone = normalize(phone_number)
+        db_phone = normalize_db_phone(phone_number)
+        reporter_phone = normalize(reporter_phone) if reporter_phone else None
+        transfer_content = (
+            normalize(transfer_content)
+            if transfer_type == 1 and transfer_content
+            else transfer_content
+        )
+        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+        try:
+            with self._engine.begin() as conn:
+                phone_row = conn.execute(
+                    text(
+                        "SELECT phone_number, tags FROM phone WHERE phone_number IN (:phone_db, :phone_full)"
+                    ),
+                    {"phone_db": db_phone, "phone_full": normalized_phone},
+                ).mappings().first()
+
+                if phone_row and phone_row["tags"] == "White":
+                    raise ValueError("該號碼已存在於白名單，無法回報為可疑號碼。")
+
+                if phone_row:
+                    conn.execute(
+                        text(
+                            "UPDATE phone SET tags = 'Black' WHERE phone_number IN (:phone_db, :phone_full)"
+                        ),
+                        {"phone_db": db_phone, "phone_full": normalized_phone},
+                    )
+                else:
+                    conn.execute(
+                        text(
+                            "INSERT INTO phone(phone_number, tags) VALUES (:phone_number, 'Black')"
+                        ),
+                        {"phone_number": db_phone},
+                    )
+
+                blacklist_row = conn.execute(
+                    text(
+                        "SELECT report_count, dominant_tag FROM blacklist WHERE phone_number IN (:phone_db, :phone_full)"
+                    ),
+                    {"phone_db": db_phone, "phone_full": normalized_phone},
+                ).mappings().first()
+
+                if blacklist_row:
+                    conn.execute(
+                        text(
+                            "UPDATE blacklist "
+                            "SET report_count = report_count + 1, last_seen_at = :last_seen_at, "
+                            "dominant_tag = :phone_type, referral_type = :referral_type, referral_info = :referral_info "
+                            "WHERE phone_number = :phone_number"
+                        ),
+                        {
+                            "last_seen_at": now,
+                            "phone_type": phone_type,
+                            "referral_type": transfer_type or 0,
+                            "referral_info": transfer_content,
+                            "phone_number": db_phone,
+                        },
+                    )
+                    total_reports = int(blacklist_row["report_count"]) + 1
+                else:
+                    conn.execute(
+                        text(
+                            "INSERT INTO blacklist(phone_number, report_count, first_seen_at, last_seen_at, dominant_tag, referral_type, referral_info) "
+                            "VALUES (:phone_number, 1, :first_seen_at, :last_seen_at, :phone_type, :referral_type, :referral_info)"
+                        ),
+                        {
+                            "phone_number": db_phone,
+                            "first_seen_at": now,
+                            "last_seen_at": now,
+                            "phone_type": phone_type,
+                            "referral_type": transfer_type or 0,
+                            "referral_info": transfer_content,
+                        },
+                    )
+                    total_reports = 1
+
+                conn.execute(
+                    text(
+                        "INSERT INTO fraud_reports(phone_number, reporter_phone, report_time, tag, referral_type, referral_info) "
+                        "VALUES (:phone_number, :reporter_phone, :report_time, :tag, :referral_type, :referral_info)"
+                    ),
+                    {
+                        "phone_number": db_phone,
+                        "reporter_phone": reporter_phone and normalize_db_phone(reporter_phone),
+                        "report_time": now,
+                        "tag": phone_type,
+                        "referral_type": transfer_type or 0,
+                        "referral_info": transfer_content,
+                    },
+                )
+        except OperationalError as exc:
+            raise RuntimeError(f"資料庫連線失敗：{exc}") from exc
+
+        return {
+            "phone_number": normalized_phone,
+            "status": "black",
+            "total_reports": total_reports,
+            "report_time": now,
+        }
+
+
+phone_repository = PhoneRepository()

backend/repository/rag_repository.py

@@ -0,0 +1,70 @@
+from pathlib import Path
+from typing import Any
+
+from backend.core.config import settings
+from backend.rag.rag_context import RAGContext
+
+
+def _require_langchain() -> tuple[Any]:
+    from langchain_chroma import Chroma
+
+    return (Chroma,)
+
+
+def get_vectorstore() -> Any:
+    """取得向量資料庫實例"""
+    (Chroma,) = _require_langchain()
+    return Chroma(
+        persist_directory=settings.RAG_PERSIST_DIR,
+        embedding_function=RAGContext.get_embeddings(),
+    )
+
+
+def query_documents(query_text: str, k: int | None = None) -> list[Any]:
+    """查詢向量庫中相似的文檔
+    
+    Args:
+        query_text: 查詢文本
+        k: 返回結果數量，默认使用配置中的 RAG_TOP_K
+    
+    Returns:
+        相似文檔列表
+    """
+    if k is None:
+        k = settings.RAG_TOP_K
+    
+    vectorstore = get_vectorstore()
+    retriever = vectorstore.as_retriever(search_kwargs={"k": k})
+    return retriever.invoke(query_text)
+
+
+def add_documents(texts: list[str], metadatas: list[dict[str, Any]] | None = None) -> list[str]:
+    """新增文檔到向量庫
+    
+    Args:
+        texts: 文本列表
+        metadatas: 元數據列表
+    
+    Returns:
+        新增文檔的 ID 列表
+    """
+    vectorstore = get_vectorstore()
+    return vectorstore.add_texts(texts=texts, metadatas=metadatas)
+
+
+def delete_documents(ids: list[str]) -> bool:
+    """從向量庫刪除文檔
+    
+    Args:
+        ids: 文檔 ID 列表
+    
+    Returns:
+        是否刪除成功
+    """
+    vectorstore = get_vectorstore()
+    try:
+        vectorstore.delete(ids=ids)
+        return True
+    except Exception as e:
+        print(f"Error deleting documents: {e}")
+        return False

backend/services/dto/price_analysis.py

@@ -0,0 +1,150 @@
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from backend.services.image_price_service.domain.models import (
+    MarketplaceCondition,
+    MarketplaceLayout,
+)
+from backend.services.image_price_service.domain.policy import (
+    DEFAULT_PRICE_RISK_POLICY,
+)
+
+RiskLabel = Literal["LOW", "MEDIUM", "HIGH", "UNKNOWN"]
+DecisionLayer = Literal["fast", "llm", "llm_simulated", "decision_error"]
+MarketPriceSource = Literal["online", "fallback_local", "not_evaluated"]
+SearchTool = Literal["serp_api", "tavily", "ddgs"]
+MarketPriceStatus = Literal["success", "insufficient", "not_found"]
+MarketPriceReferenceMode = Literal["iqr", "median_low_sample"]
+
+
+class MarketPriceCandidateEvidence(BaseModel):
+    """保留已驗證市場候選的來源與價格證據。"""
+
+    model_config = ConfigDict(extra="forbid")
+
+    candidate_id: str
+    title: str
+    price: int = Field(gt=0)
+    condition: MarketplaceCondition
+    evidence: str
+
+
+class MarketPriceEstimate(BaseModel):
+    """保留市場價格區間、資料量、來源與可信度。"""
+
+    model_config = ConfigDict(extra="forbid")
+
+    status: MarketPriceStatus
+    condition: MarketplaceCondition
+    reference_mode: MarketPriceReferenceMode
+    median_price: int = Field(ge=0)
+    low_price: int = Field(ge=0)
+    high_price: int = Field(ge=0)
+    sample_count: int = Field(ge=0)
+    site_count: int = Field(ge=0)
+    source: MarketPriceSource
+    confidence: float = Field(ge=0.0, le=1.0)
+    search_tools: list[SearchTool] = Field(default_factory=list)
+    candidates: tuple[MarketPriceCandidateEvidence, ...] = ()
+
+
+class DeepAnalysisReview(BaseModel):
+    """LLM 對有限商品狀態原文所做的結構化複核。"""
+
+    model_config = ConfigDict(extra="forbid")
+
+    reviewed_condition: MarketplaceCondition
+    condition_detail: str = ""
+    condition_evidence: str = ""
+    reason: str
+    review_confidence: float = Field(ge=0.0, le=1.0)
+
+    @model_validator(mode="after")
+    def validate_review_consistency(self) -> "DeepAnalysisReview":
+        """拒絕缺少證據或與 UNKNOWN 狀態矛盾的複核結果。"""
+        if (
+            self.reviewed_condition != MarketplaceCondition.UNKNOWN
+            and not self.condition_evidence.strip()
+        ):
+            raise ValueError("確認商品狀態時必須提供可追溯的狀態原文證據")
+
+        if (
+            self.reviewed_condition == MarketplaceCondition.UNKNOWN
+            and self.condition_detail.strip()
+        ):
+            raise ValueError("商品狀態為 UNKNOWN 時不得提供已確認的狀態細節")
+
+        return self
+
+
+class ImagePriceAnalysisResult(BaseModel):
+    """圖片價格分析的內部完整結果"""
+
+    model_config = ConfigDict(extra="forbid")
+
+    filename: str
+    content_type: str
+    extracted_text: str
+    success: bool = True
+    error_code: str | None = None
+    message: str | None = None
+    product_name: str | None = None
+    brand_model: str | None = None
+    listed_price: int | None = Field(default=None, ge=0)
+    market_price: int = Field(default=0, ge=0)
+    market_price_source: MarketPriceSource = "not_evaluated"
+    market_price_estimates: tuple[MarketPriceEstimate, ...] = ()
+    risk_label: RiskLabel = "UNKNOWN"
+    score: str | float | None = None
+    reason: str | None = None
+    evidence: list[str] = Field(default_factory=list)
+    confidence: float | None = Field(default=None, ge=0, le=1)
+    decision_layer: DecisionLayer = "fast"
+    search_tools: list[SearchTool] = Field(default_factory=list)
+    marketplace_layout: MarketplaceLayout = MarketplaceLayout.UNKNOWN
+    marketplace_confidence: float = Field(default=0.0, ge=0, le=1)
+    extraction_confidence: float | None = Field(default=None, ge=0, le=1)
+    price_source_text: str | None = None
+    price_extraction_reason: str | None = None
+    seller_name: str | None = None
+    condition: MarketplaceCondition = MarketplaceCondition.UNKNOWN
+    condition_detail: str = ""
+    condition_source_text: str = ""
+    condition_extraction_confidence: float = Field(
+        default=0.0,
+        ge=0.0,
+        le=1.0,
+        description=(
+            "只代表 extractor 擷取商品狀態的可信度，不代表商品匹配、"
+            "市場資料或價格風險可信度"
+        ),
+    )
+    extraction_warnings: list[str] = Field(default_factory=list)
+
+    @field_validator("listed_price")
+    @classmethod
+    def validate_supported_listing_price(cls, value: int | None) -> int | None:
+        """拒絕超出服務支援上限的刊登價格。"""
+        if (
+            value is not None
+            and value > DEFAULT_PRICE_RISK_POLICY.maximum_supported_price
+        ):
+            raise ValueError(
+                "PRICE_OUT_OF_SUPPORTED_RANGE：刊登價格超出支援範圍"
+            )
+        return value
+
+
+class ProductIdentification(BaseModel):
+    """商品辨識階段輸出的正規化資訊、查價搜尋詞與本地參考價格。"""
+
+    product_name: str
+    brand_model: str
+    known_specs: list[str] = Field(default_factory=list)
+    search_query: str = ""
+    market_price: int = Field(default=0, ge=0)
+
+
+class OCRResponse(BaseModel):
+    extracted_text: str = Field(..., description="OCR 萃取文字")

backend/services/dto/text_analysis.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+
+
+class BaseEvidence(BaseModel):
+    text: str
+    rule_score: int
+    rule_reason: str
+    rule_hits: list[str]
+    model_label: str
+    model_confidence: float | None
+    model_margin: float | None
+    chunk_consistent: bool 

backend/services/image_price_service/__init__.py

@@ -0,0 +1 @@
+

backend/services/image_price_service/case_recorder.py

@@ -0,0 +1,58 @@
+"""將圖片價格分析結果轉成案例資料並以 best-effort 方式保存。"""
+
+import logging
+from typing import Any
+
+from backend.core.config import settings
+from backend.services.dto.price_analysis import ImagePriceAnalysisResult
+
+logger = logging.getLogger(__name__)
+
+
+def record_case(
+    case_repo: Any,
+    result: ImagePriceAnalysisResult,
+) -> str | None:
+    """依設定保存分析案例；寫入失敗時維持原分析結果。"""
+    if not settings.CASE_MEMORY_ENABLED:
+        return None
+
+    try:
+        return case_repo.save(
+            {
+                "filename": result.filename,
+                "content_type": result.content_type,
+                "product_name": result.product_name,
+                "brand_model": result.brand_model,
+                "selling_price": result.listed_price,
+                "market_price": result.market_price,
+                "market_price_source": result.market_price_source,
+                "market_price_estimates": [
+                    estimate.model_dump(exclude={"candidates"}, mode="json")
+                    for estimate in result.market_price_estimates
+                ],
+                "risk_label": result.risk_label,
+                "risk_score": result.score,
+                "reason": result.reason,
+                "evidence": result.evidence,
+                "confidence": result.confidence,
+                "decision_layer": result.decision_layer,
+                "error_code": result.error_code,
+                "extracted_text": result.extracted_text,
+                "marketplace_layout": result.marketplace_layout,
+                "marketplace_confidence": result.marketplace_confidence,
+                "price_source_text": result.price_source_text,
+                "price_extraction_reason": result.price_extraction_reason,
+                "seller_name": result.seller_name,
+                "condition": result.condition,
+                "condition_detail": result.condition_detail,
+                "condition_source_text": result.condition_source_text,
+                "condition_extraction_confidence": (
+                    result.condition_extraction_confidence
+                ),
+                "extraction_warnings": result.extraction_warnings,
+            }
+        )
+    except Exception:
+        logger.warning("圖片價格分析案例保存失敗", exc_info=True)
+        return None

backend/services/image_price_service/domain/__init__.py

@@ -0,0 +1 @@
+"""商品價格驗證的 domain models 與 policy。"""

backend/services/image_price_service/domain/models.py

@@ -0,0 +1,152 @@
+"""定義 OCR、商品頁驗證與刊登欄位抽取共用的資料模型。"""
+
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel, Field, model_validator
+
+from backend.services.image_price_service.domain.policy import (
+    DEFAULT_PRICE_RISK_POLICY,
+)
+
+
+class MarketplaceCondition(str, Enum):
+    """刊登商品的使用狀況。"""
+
+    NEW = "new"
+    USED = "used"
+    UNKNOWN = "unknown"
+
+
+class MarketplaceLayout(str, Enum):
+    """商品頁截圖的版面類型。"""
+
+    MOBILE = "mobile"
+    DESKTOP = "desktop"
+    UNKNOWN = "unknown"
+
+
+class PriceSection(str, Enum):
+    """價格候選值所在的商品頁區段。"""
+
+    MAIN_PRICE = "main_price"
+    OFFER_RANGE = "offer_range"
+    DESCRIPTION = "description"
+    SELLER_INFO = "seller_info"
+    MESSAGE_BOX = "message_box"
+    DETAIL = "detail"
+    UNKNOWN = "unknown"
+
+
+class OCRTextBlock(BaseModel):
+    """保存單一 OCR 文字區塊及其座標。"""
+
+    text: str
+    x: float | None = None
+    y: float | None = None
+    width: float | None = None
+    height: float | None = None
+    confidence: float | None = None
+
+
+class OCRDocument(BaseModel):
+    """保存 OCR 全文、文字區塊與頁面尺寸。"""
+
+    text: str
+    blocks: list[OCRTextBlock] = Field(default_factory=list)
+    width: float | None = None
+    height: float | None = None
+
+    @property
+    def has_coordinates(self) -> bool:
+        """判斷所有文字區塊是否都有座標。"""
+        return bool(self.blocks) and all(
+            block.x is not None and block.y is not None
+            for block in self.blocks
+        )
+
+
+class ProductAgentResult(BaseModel):
+    """保存模型輸出、搜尋結果與工具錯誤。"""
+
+    output: dict[str, Any]
+    tool_results: list[dict[str, Any]]
+    tool_errors: list[str]
+
+class DetectionResult(BaseModel):
+    """商品頁來源驗證與版型推定結果。"""
+
+    is_marketplace: bool
+    layout: MarketplaceLayout
+    confidence: float
+    evidence: list[str] = Field(default_factory=list)
+    reason: str | None = None
+
+
+class PriceCandidate(BaseModel):
+    """從 OCR 文字擷取的單一價格候選及其判定資訊。"""
+
+    amount: int
+    currency: str
+    source_text: str
+    block_index: int
+    x: float | None
+    y: float | None
+    context_before: str | None
+    context_after: str | None
+    section: PriceSection
+    confidence: float
+    reject_reason: str | None = None
+
+
+class MainPriceExtractionResult(BaseModel):
+    """保存 Marketplace 刊登欄位的抽取結果。"""
+
+    price: int | None
+    currency: str | None
+    confidence: float
+    source_text: str | None
+    layout: MarketplaceLayout
+    candidates: list[PriceCandidate] = Field(default_factory=list)
+    rejected_candidates: list[PriceCandidate] = Field(default_factory=list)
+    error_code: str | None = None
+    message: str | None = None
+    reason: str | None = None
+    product_name: str | None = None
+    seller_name: str | None = None
+    condition: MarketplaceCondition = MarketplaceCondition.UNKNOWN
+    condition_detail: str = ""
+    condition_source_text: str = ""
+    condition_extraction_confidence: float = Field(
+        default=0.0,
+        ge=0.0,
+        le=1.0,
+        description=(
+            "只代表 extractor 擷取商品狀態的可信度，不代表商品匹配、"
+            "市場資料或價格風險可信度"
+        ),
+    )
+    warnings: list[str] = Field(default_factory=list)
+
+    @model_validator(mode="after")
+    def validate_supported_listing_price(self) -> "MainPriceExtractionResult":
+        """將超出服務上限的刊登價格標記為明確的擷取錯誤。"""
+        if (
+            self.price is not None
+            and self.price
+            > DEFAULT_PRICE_RISK_POLICY.maximum_supported_price
+        ):
+            self.error_code = "PRICE_OUT_OF_SUPPORTED_RANGE"
+            self.message = "刊登價格超出商品價格驗證服務支援範圍"
+        return self
+
+class MainPriceExtractionError(ValueError):
+    """表示主價格不存在或擷取信心不足。"""
+
+    def __init__(self, result: MainPriceExtractionResult):
+        """建立錯誤並保留原始擷取結果。"""
+        if not result.error_code:
+            raise ValueError("MainPriceExtractionError 必須包含 error_code")
+        self.result = result
+        self.error_code = result.error_code
+        super().__init__(result.message or result.error_code)