# 通过 Agent 调用 result = agent. invoke ({     "messages" : [{ "role" : "user" , "content" : "获取 sandbox 信息" }] }) # 或直接通过管理器获取 manager = get_sandbox_manager () info = manager. get_info ()

import webbrowser import urllib. parse from pathlib import Path def open_vnc_viewer ( vnc_url : str):     "" "自动打开 VNC 查看器" ""    current_dir = Path (__file__). parent . absolute ()    vnc_html_path = current_dir / "vnc.html"         if vnc_html_path. exists ():        # 通过 URL 参数传递 VNC URL        encoded_url = urllib. parse . quote (vnc_url, safe= '' )        file_url = f " file://{vnc_html_path}?url={encoded_url}"

// 从 URL 参数获取 VNC URL const urlParams = new URLSearchParams ( window . location . search ); const vncUrl = urlParams. get ( 'url' ); // 加载 noVNC 库 async function loadNoVNC ( ) {     const module = await import ( 'https://cdn.jsdelivr.net/gh/novnc/noVNC@v1.4.0/core/rfb.js' );     return module . default ; } // 连接 VNC async function connectVNC ( url ) {     const RFB = await loadNoVNC ();    rfb = new RFB (vncScreen, url, {         shared : true ,         credentials : { password : '' }    });        rfb. addEventListener ( 'connect' , () => {         console . log ( 'VNC 连接成功' );

遇到的问题

• include权限限制：私有仓库的remote include需要特殊权限配置，导致"Invalid configuration format"错误。
• 访问控制：即使启用"Allow anyone to pull from package registry"，CI include仍受项目可见性限制。
• 维护复杂性：依赖外部文件增加了配置的脆弱性。

解决方案：直接嵌入CI作业

• 设计调整：放弃include机制，将共享CI作业直接复制到各项目的  .gitlab-ci.yml 中。
• 优势：
  • 消除权限依赖，确保各项目独立运行。
  • 简化配置，减少外部依赖。
  • 保持代码复用（通过共享包），同时保证CI配置的自主性。
• 迁移策略：逐步替换各项目的include为嵌入式配置，确保平滑过渡。

更新引用项目

• 示例项目1：最初测试include方式，后因权限问题转为嵌入。
• 示例项目2：master和OTA4分支均采用嵌入配置。
• 示例项目3：OTA2分支嵌入CI作业。

迁移过程

1. 从单项目脚本迁移：提取公共逻辑到共享包。
2. 测试include方案：验证包安装和CI触发。
3. 识别权限问题：分析失败原因，调整为嵌入方案。
4. 多项目部署：逐个更新项目的CI配置，确保一致性。

variables:   PYPI_INDEX_URL: 'https://gitlab.example.com/api/v4/projects/12345/packages/pypi/simple' ai_code_review:   stage: review   tags:     - ai-review   rules:     - if: $CI_MERGE_REQUEST_IID     - if: $CI_PIPELINE_SOURCE == "push"   variables:     GIT_DEPTH: 1   timeout: 30m   before_script:     - pip install ai-code-review-tool --index-url $PYPI_INDEX_URL   script:     -   kkl.zw0769.com  |      if [ -n "$CI_MERGE_REQUEST_IID" ]; then        ai-review --project-id $CI_PROJECT_ID --mr-iid $CI_MERGE_REQUEST_IID --gitlab-url $CI_SERVER_URL --token $GITLAB_TOKEN      else        ai-review --project-id $CI_PROJECT_ID --gitlab-url $CI_SERVER_URL --token $G

 <dubbo.version> 3.0 .3 </dubbo.version> </properties> <dependencies>    <dependency>        <groupId>org.apache.dubbo</groupId>        <artifactId>dubbo</artifactId>        <version>${dubbo.version}</version>    </dependency>    <dependency>        <groupId>org.apache.dubbo</groupId>        <artifactId>dubbo-dependencies-zookeeper</artifactId>        <version>${dubbo.version}</version>

# models.py from sqlalchemy import Column, Integer, String, create_engine from sqlalchemy.ext.declarative import declarative_base DATABASE_URL = "sqlite:///./test.db" # 注意：`check_same_thread=False` 仅用于SQLite简化示例，生产环境需用更安全的方式 engine = create_engine( DATABASE_URL , connect_args={ "check_same_thread" : False }) Base = declarative_base() class Item( Base):    __tablename__ = "items"    id = Column( Integer, primary_key= True, index= True)    name = Column( String, index= True)    description = Column( String, nullable= True) # 创建表 Base.metadata.create_all( bind= engine)  xpg.zw0769.com

from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.orm import Session from . import models from .database import SessionLocal, engine app = FastAPI() # 依赖项：获取数据库会话 def get_db ():mdw.zw0769.com    db = SessionLocal()     try :         yield db     finally :        db.close() @app.post( "/items/") def create_item ( name: str, description: str = None, db: Session = Depends( get_db) ):    db_item = models.Item(name=name, description=description)    db.add(db_item)    db.commit()    db.refresh(db_item)     return db_item @app.get( "/items/{item_id}") def read_item ( item_id: int, db: Session = Depends( get_db) ):    item = db.query(models.Item). filter (models.Item. id == item_id).first()     if item is None :

from jose import JWTError, jwt from datetime import datetime, timedelta SECRET_KEY = "your-secret-key-change-this-in-production" # 🔴 生产环境必须用强密钥且从环境变量读取！ ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 def create_access_token ( data: dict ):    to_encode = data.copy()hce.zw0769.com    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)    to_encode.update({ "exp" : expire})    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)     return encoded_jwt async def get_current_user ( token: str = Depends( oauth2_scheme), db: Session = Depends( get_db) ):    credentials_exception = HTTPException(status_code= 401 , detail= "无效凭证" )     try :        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])        username: str = payload.get( "sub" )         if username is None :             raise credentials_exception     except JWTError:         raise credentials_exception    user = db.query(models.User). filter (models.User.username == username).first()

# database_async.py from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker # 使用aiosqlite驱动 ASYNC_DATABASE_URL = "sqlite+aiosqlite:///./test_async.db" async_engine = create_async_engine(ASYNC_DATABASE_URL, pool_pre_ping= True , pool_size= 10 , max_overflow= 20 ) AsyncSessionLocal = sessionmaker(hfg.zw0769.com    bind=async_engine,    class_=AsyncSession,    expire_on_commit= False ) async def get_async_db ():     async with AsyncSessionLocal() as session:         yield session # 在FastAPI路由中使用 @app.post( "/async-items/", response_model=schemas.Item) async def create_async_item (    item: schemas.ItemCreate,    db: AsyncSession = Depends( get_async_db) ):    db_item = models.Item(**item. dict ())  wjk.zw0769.com

your_project/ ├── main.py           # FastAPI应用创建和路由聚合 ├── models.py         # SQLAlchemy数据模型 ├── schemas.py       # Pydantic请求/响应模型（用于数据验证） ├── crud.py           # 核心的数据库操作函数 ├── database.py       # 数据库引擎和会话工厂（同步/异步） ├── auth.py           # 认证、令牌相关函数 ├── security.py       # 密码哈希函数

lil.zw0769.com

from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from . import models, schemas, crud, auth from .database import AsyncSessionLocal, get_async_db app = FastAPI(title= "My Secure API" ) # 添加CORS中间件（根据需求配置） app.add_middleware(CORSMiddleware, allow_origins=[ "*" ]) # 生产环境应指定具体域名 @app.post( "/token") async def login_for_access_token ( form_data: OAuth2PasswordRequestForm = Depends( ), db: AsyncSession = Depends( get_async_db) ):    user = await authenticate_user(form_data.username, form_data.password, db)     if not user:         raise HTTPException(status_code= 400 , detail= "用户名或密码错误" )cpnapt.com    access_token = auth.create_access_token(data={ "sub" : user.username})     return { "access_token" : access_token, "token_type" : "bearer" } @app.get( "/users/me/") async def read_users_me ( current_user: models.User = Depends( auth.get_current_user) ):  dgz.zw0769.com     return current_user # 受保护的、异步的、带后台任务的路由示例 @app.post( "/items-with-notify/") async def create_item_notify (jrs.cpnapt.com    item: schemas.ItemCreate,    background_tasks: BackgroundTasks,xfk.zw0769.com    db: AsyncSession = Depends( get_async_db),    current_user: models.User = Depends( auth.get_current_user) ):    db_item = await crud.create_user_item(db=db, item=item, user_id=current_user. id )     # 假设有个发送通知的函数    background_tasks.add_task(send_notification, f"新项目 {item.name} 已创建！" )