Spaces:

Stouching-Robotics
/

DataFlowing

Running

stouchinglhh

Claude

sync processor from GitLab c697b6e: 版本 1.19.4 / 前端 1.12.4 —— 服务端 i18n(首屏不再闪英文)、推理模块并入模型页、前端抽出 shared/、缓存键改用前端号

bb59211 about 1 month ago

6.87 kB

"""FastAPI application entry point — 纯本地文件存储,不依赖 PostgreSQL。"""

import asyncio
import mimetypes
import threading
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from starlette.middleware.gzip import GZipMiddleware

from app.config import settings
from app.routes import ingestion, video, export, pages, session, annotations, dashboard, devices, auth
from app.routes import training
from app.api import workflows, worker, projects, exceptions, quality
from app.api import users
from app.i18n import LanguageMiddleware
from app.middleware import AuthMiddleware
from app.paths import STATIC_DIR
from app.version import __version__


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Metadata remains file-backed for compatibility, while EGOData stores
    # managed accounts and future relational metadata.
    try:
        from app.database import init_db
        await init_db()
    except Exception as exc:
        # Keep the legacy file-backed application available during a temporary
        # database/tunnel outage; auth.py will use its JSON fallback.
        print(f"[Startup] Database initialization skipped: {exc}")
    from app import localstore
    localstore.STATE_ROOT.mkdir(parents=True, exist_ok=True)
    try:
        if localstore.migrate_workflow_definitions():
            print("[Startup] Migrated workflow node types to the canonical RGB/RGB-D schema")
    except Exception as exc:
        # The API/worker also accept legacy aliases, so a migration failure is
        # non-fatal and can be retried on the next startup.
        print(f"[Startup] Workflow type migration skipped: {exc}")
    # Do the one potentially slow storage scan before accepting requests.  This
    # keeps a cold network-mounted data directory from becoming the first page
    # load seen by a user after a service restart.  A transient storage issue
    # must not prevent the compatibility service from starting.
    try:
        await asyncio.to_thread(localstore.scan_sessions)
    except Exception as exc:
        print(f"[Startup] Session cache warm-up skipped: {exc}")
    threading.Thread(
        target=localstore.warm_metadata_caches,
        name="metadata-cache-warmup",
        daemon=True,
    ).start()
    from app import upload_queue
    upload_queue.start()
    print(f"[Startup] processor Service on port {settings.PORT}")
    print(f"[Startup] Storage: {settings.STORAGE_DIR} ({settings.STORAGE_BACKEND})")
    try:
        yield
    finally:
        upload_queue.stop()
        print("[Shutdown] Done")


# 版本号取自 app/version.py —— 此前这里硬编码 "0.1.0",与 version.py 的
# 1.7.0 长期不一致,/docs 的 OpenAPI 版本与 /health 报的是两个数。
app = FastAPI(
    title="processor Service",
    version=__version__,
    lifespan=lifespan,
)

# GZip compression — JSON responses shrink ~10×
app.add_middleware(GZipMiddleware, minimum_size=500)

# Auth middleware — protect routes (after gzip, before everything else)
# 语言中间件要在 Auth 之外(先于它)跑:鉴权失败跳登录页时,那个页面本身也要
# 按 cookie 里的语言渲染。add_middleware 是**后加的先跑**,所以放在 Auth 之后加。
app.add_middleware(AuthMiddleware)
app.add_middleware(LanguageMiddleware)

# 媒体/压缩文件下载不走 gzip:zip/mp4/parquet/hdf5 本身已是压缩格式,
# 再压一遍只是烧 CPU(zlib level 9 压缩 180MB ≈ 10s),而且流式压缩会
# 删掉 Content-Length —— 浏览器下载进度显示"未知大小",大文件看起来
# 像卡死。这些路径直接透传原始字节(去掉 accept-encoding 头即可)。
_NO_COMPRESS_PATH_HINTS = (
    "download", "stream", "depth-codes", "depth-preview", "skeleton", "hdf5",
    "tactile-matrix-data",
)


@app.middleware("http")
async def no_compress_media(request, call_next):
    if any(hint in request.url.path for hint in _NO_COMPRESS_PATH_HINTS):
        request.scope["headers"] = [
            (key, value) for key, value in request.scope["headers"]
            if key.lower() != b"accept-encoding"
        ]
    return await call_next(request)

# Static JS/CSS is versioned (workflow assets use content hashes; the legacy
# page scripts use a ``?v=...`` query).  Tell the browser to retain these
# resources across full page navigation instead of revalidating every common
# asset on each switch between Projects, Review and Workflow Studio.
@app.middleware("http")
async def static_cache_headers(request, call_next):
    response = await call_next(request)
    path = request.url.path
    if path.startswith("/static/"):
        if "/assets/" in path and "/workflow-studio/" in path:
            response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
        elif request.url.query:
            response.headers["Cache-Control"] = "public, max-age=86400"
        elif path.startswith("/static/js/"):
            # ES module 的相对 import 解析出来的 URL **不带查询串** —— 就算入口
            # script 写了 ?v=,`import './api.js'` 也永远是同一个地址。给它任何
            # 正数的 max-age,升级后都会出现「新入口 + 旧子模块」混用,直接抛
            # TypeError。这类文件必须每次回源校验(未变则 304,开销可忽略)。
            #
            # 范围刻意收窄到 /static/js/:带 ?v= 的老脚本走上面那支不受影响;
            # /static/ 顶层的第三方库(plyr.js、iconify-*)也不在范围里。
            response.headers["Cache-Control"] = "no-cache"
        else:
            response.headers["Cache-Control"] = "public, max-age=3600"
    return response

# Static files (CSS/JS)
static_dir = STATIC_DIR
static_dir.mkdir(parents=True, exist_ok=True)
mimetypes.add_type("text/javascript", ".js")
mimetypes.add_type("text/javascript", ".mjs")
mimetypes.add_type("text/css", ".css")
mimetypes.add_type("application/json", ".json")
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")

# Routes
app.include_router(ingestion.router)
app.include_router(video.router)
app.include_router(export.router)
app.include_router(pages.router)
app.include_router(session.router)
app.include_router(annotations.router)
from app import ai_annotation  # AI 辅助标注(信号切段 + VLM 提案)
app.include_router(ai_annotation.router)
app.include_router(dashboard.router)
# tasks.router 已并入 projects.router(项目概念,任务已移除),不再注册
app.include_router(devices.router)
app.include_router(auth.router)
app.include_router(workflows.router)
app.include_router(worker.router)
app.include_router(projects.router)
app.include_router(exceptions.router)
app.include_router(quality.router)
app.include_router(users.router)
app.include_router(training.router)

	"""FastAPI application entry point — 纯本地文件存储,不依赖 PostgreSQL。"""

	import asyncio
	import mimetypes
	import threading
	from contextlib import asynccontextmanager
	from pathlib import Path

	from fastapi import FastAPI
	from fastapi.staticfiles import StaticFiles
	from starlette.middleware.gzip import GZipMiddleware

	from app.config import settings
	from app.routes import ingestion, video, export, pages, session, annotations, dashboard, devices, auth
	from app.routes import training
	from app.api import workflows, worker, projects, exceptions, quality
	from app.api import users
	from app.i18n import LanguageMiddleware
	from app.middleware import AuthMiddleware
	from app.paths import STATIC_DIR
	from app.version import __version__


	@asynccontextmanager
	async def lifespan(app: FastAPI):
	# Metadata remains file-backed for compatibility, while EGOData stores
	# managed accounts and future relational metadata.
	try:
	from app.database import init_db
	await init_db()
	except Exception as exc:
	# Keep the legacy file-backed application available during a temporary
	# database/tunnel outage; auth.py will use its JSON fallback.
	print(f"[Startup] Database initialization skipped: {exc}")
	from app import localstore
	localstore.STATE_ROOT.mkdir(parents=True, exist_ok=True)
	try:
	if localstore.migrate_workflow_definitions():
	print("[Startup] Migrated workflow node types to the canonical RGB/RGB-D schema")
	except Exception as exc:
	# The API/worker also accept legacy aliases, so a migration failure is
	# non-fatal and can be retried on the next startup.
	print(f"[Startup] Workflow type migration skipped: {exc}")
	# Do the one potentially slow storage scan before accepting requests. This
	# keeps a cold network-mounted data directory from becoming the first page
	# load seen by a user after a service restart. A transient storage issue
	# must not prevent the compatibility service from starting.
	try:
	await asyncio.to_thread(localstore.scan_sessions)
	except Exception as exc:
	print(f"[Startup] Session cache warm-up skipped: {exc}")
	threading.Thread(
	target=localstore.warm_metadata_caches,
	name="metadata-cache-warmup",
	daemon=True,
	).start()
	from app import upload_queue
	upload_queue.start()
	print(f"[Startup] processor Service on port {settings.PORT}")
	print(f"[Startup] Storage: {settings.STORAGE_DIR} ({settings.STORAGE_BACKEND})")
	try:
	yield
	finally:
	upload_queue.stop()
	print("[Shutdown] Done")


	# 版本号取自 app/version.py —— 此前这里硬编码 "0.1.0",与 version.py 的
	# 1.7.0 长期不一致,/docs 的 OpenAPI 版本与 /health 报的是两个数。
	app = FastAPI(
	title="processor Service",
	version=__version__,
	lifespan=lifespan,
	)

	# GZip compression — JSON responses shrink ~10×
	app.add_middleware(GZipMiddleware, minimum_size=500)

	# Auth middleware — protect routes (after gzip, before everything else)
	# 语言中间件要在 Auth 之外(先于它)跑:鉴权失败跳登录页时,那个页面本身也要
	# 按 cookie 里的语言渲染。add_middleware 是后加的先跑,所以放在 Auth 之后加。
	app.add_middleware(AuthMiddleware)
	app.add_middleware(LanguageMiddleware)

	# 媒体/压缩文件下载不走 gzip:zip/mp4/parquet/hdf5 本身已是压缩格式,
	# 再压一遍只是烧 CPU(zlib level 9 压缩 180MB ≈ 10s),而且流式压缩会
	# 删掉 Content-Length —— 浏览器下载进度显示"未知大小",大文件看起来
	# 像卡死。这些路径直接透传原始字节(去掉 accept-encoding 头即可)。
	_NO_COMPRESS_PATH_HINTS = (
	"download", "stream", "depth-codes", "depth-preview", "skeleton", "hdf5",
	"tactile-matrix-data",
	)


	@app.middleware("http")
	async def no_compress_media(request, call_next):
	if any(hint in request.url.path for hint in _NO_COMPRESS_PATH_HINTS):
	request.scope["headers"] = [
	(key, value) for key, value in request.scope["headers"]
	if key.lower() != b"accept-encoding"
	]
	return await call_next(request)

	# Static JS/CSS is versioned (workflow assets use content hashes; the legacy
	# page scripts use a ``?v=...`` query). Tell the browser to retain these
	# resources across full page navigation instead of revalidating every common
	# asset on each switch between Projects, Review and Workflow Studio.
	@app.middleware("http")
	async def static_cache_headers(request, call_next):
	response = await call_next(request)
	path = request.url.path
	if path.startswith("/static/"):
	if "/assets/" in path and "/workflow-studio/" in path:
	response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
	elif request.url.query:
	response.headers["Cache-Control"] = "public, max-age=86400"
	elif path.startswith("/static/js/"):
	# ES module 的相对 import 解析出来的 URL 不带查询串 —— 就算入口
	# script 写了 ?v=,`import './api.js'` 也永远是同一个地址。给它任何
	# 正数的 max-age,升级后都会出现「新入口 + 旧子模块」混用,直接抛
	# TypeError。这类文件必须每次回源校验(未变则 304,开销可忽略)。
	#
	# 范围刻意收窄到 /static/js/:带 ?v= 的老脚本走上面那支不受影响;
	# /static/ 顶层的第三方库(plyr.js、iconify-*)也不在范围里。
	response.headers["Cache-Control"] = "no-cache"
	else:
	response.headers["Cache-Control"] = "public, max-age=3600"
	return response

	# Static files (CSS/JS)
	static_dir = STATIC_DIR
	static_dir.mkdir(parents=True, exist_ok=True)
	mimetypes.add_type("text/javascript", ".js")
	mimetypes.add_type("text/javascript", ".mjs")
	mimetypes.add_type("text/css", ".css")
	mimetypes.add_type("application/json", ".json")
	app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")

	# Routes
	app.include_router(ingestion.router)
	app.include_router(video.router)
	app.include_router(export.router)
	app.include_router(pages.router)
	app.include_router(session.router)
	app.include_router(annotations.router)
	from app import ai_annotation # AI 辅助标注(信号切段 + VLM 提案)
	app.include_router(ai_annotation.router)
	app.include_router(dashboard.router)
	# tasks.router 已并入 projects.router(项目概念,任务已移除),不再注册
	app.include_router(devices.router)
	app.include_router(auth.router)
	app.include_router(workflows.router)
	app.include_router(worker.router)
	app.include_router(projects.router)
	app.include_router(exceptions.router)
	app.include_router(quality.router)
	app.include_router(users.router)
	app.include_router(training.router)