ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 人工智能 > 深度学习 > HagiCode 项目使用 Docker 镜像作为核心运行时组件

HagiCode 项目使用 Docker 镜像作为核心运行时组件

投稿 深度学习 作者：ecclw459 时间：2026-01-26 16:05:22 0 删除 编辑

为解决这些问题，我们需要建立一个自动化的镜像同步机制，将 Docker Hub 的镜像定期同步到 Azure ACR，确保用户能够在 Azure 环境中获得更快的镜像拉取速度和更高的可用性。

关于 HagiCode

我们正在开发 HagiCode——一款 AI 驱动的代码智能助手，让开发体验变得更智能、更便捷、更有趣。

智能——AI 全程辅助，从想法到代码，让编码效率提升数倍。便捷——多线程并发操作，充分利用资源，开发流程顺畅无阻。有趣——游戏化机制和成就系统，让编码不再枯燥，充满成就感。

项目正在快速迭代中，如果你对技术写作、知识管理或者 AI 辅助开发感兴趣，欢迎来 GitHub 看看。

技术方案对比

在制定解决方案时，我们对比了多种技术方案：

1. image-syncer（最终选择）

  • 增量同步：仅同步变更的镜像层，显著减少网络传输
  • 断点续传：网络中断后可恢复同步
  • 并发控制：支持配置并发线程数，提升大镜像同步效率
  • 完善的错误处理：内置失败重试机制（默认 3 次）
  • 轻量级部署：单二进制文件，无依赖
  • 多仓库支持：兼容 Docker Hub、Azure ACR、Harbor 等

2. Docker CLI

  • 不支持增量同步：每次都需要拉取完整的镜像内容
  • 效率较低：网络传输量大，时间长
  • 简单易用：使用熟悉的 docker pull/push 命令

 schedule:

    - cron: "0 0 * * *" # 每天 UTC 00:00

  workflow_dispatch: # 手动触发

  push:

    branches: [publish]

 

permissions:

  contents: read

 

jobs:

  sync:

    runs-on: ubuntu-latest

 

    steps:

      - name: Checkout code

        uses: actions/checkout@v4

 

      - name: Download image-syncer

        run: |

          # 下载 image-syncer 二进制文件

          wget https://github.com/AliyunContainerService/image-syncer/releases/download/v1.5.5/image-syncer-v1.5.5-linux-amd64.tar.gz

          tar -zxvf image-syncer-v1.5.5-linux-amd64.tar.gz

          chmod +x image-syncer

 

      - name: Create auth config

        run: |

          # 生成认证配置文件 (YAML 格式)

          cat > auth.yaml <<EOF

          hagicode.azurecr.io:

            username: "${{ secrets.AZURE_ACR_USERNAME }}"

            password: "${{ secrets.AZURE_ACR_PASSWORD }}"

          EOF

 

      - name: Create images config

        run: |

          # 生成镜像同步配置文件 (YAML 格式)

          cat > images.yaml <<EOF

          docker.io/newbe36524/hagicode: hagicode.azurecr.io/hagicode

          EOF

 

      - name: Run image-syncer

        run: |

          # 执行同步 (使用新版 --auth 和 --images 参数)

          ./image-syncer --auth=./auth.yaml --images=./images.yaml --proc=10 --retries=3

 

      - name: Upload logs

        if: always()

        uses: actions/upload-artifact@v4

        with:

          name: sync-logs

          path: image-syncer-*.log

          retention-days: 7

配置说明

1. 触发条件

定时触发：cron: "0 0 * * *" - 每天 UTC 00:00 执行

手动触发：workflow_dispatch - 允许用户在 GitHub UI 手动运行

推送触发：push: branches: [publish] - 发布分支推送时触发（用于快速同步）


原有脚本项目核心目录保留不变，改造后新增「API 层、配置层」，确保原有脚本无需修改即可复用。

  • 改造前目录（即脚本项目原始目录）示例如下：

    SUPER-API-AUTO-TEST/          # 接口自动化测试项目根目录├── auth.py                    # 鉴权相关├── case_generator.py          # 用例生成逻辑├── config.yaml                # 项目配置├── conftest.py                # Pytest夹具├── runner.py                  # 用例执行入口├── reports/                   # 测试报告目录├── logs/                      # 日志目录├── testcases/                 # 测试用例脚本（Python）├── testcases_data/            # 测试用例数据（YAML）└── utils/                     # 通用工具类

  • Fastapi 改造后目录结构示例如下：

    FASTAPI-SUPER-API-AUTO-TEST/  # 基于FastAPI的改造后目录├── auth.py                    
├── case_generator.py          
├── conftest.py                
├── main.py                   # FastAPI应用入口（核心新增）├── pytest.ini                
├── runner.py                  
├── api/                      # API层（核心新增）│   ├── testcase_route.py     # 接口路由定义（URL、请求参数）│   ├── testcase_service.py   # 接口业务逻辑（与原有脚本交互）│   └── __init__.py           
├── configs/                  # 配置目录（拆分原有config.yaml）│   └── config.yaml           
├── logs/                     
├── reports/                  
├── testcases/                # 完全复用原有目录├── testcases_data/           # 完全复用原有目录└── utils/                    # 完全复用原有目录

改造后，新增了 Fastapi 路由层  api/、配置层  configs/，以及 FastAPI 应用入口  main.py。

三. 核心接口设计与实现

1. 接口设计示例

我们设计了以下关键接口供测试平台调用：

① 获取项目与模块信息

  • GET /api_test/testcases/projects：获取所有项目列表
  • GET /api_test/testcases/modules：获取指定项目下的模块列表

② 获取用例列表

  • GET /api_test/testcases/list：支持按项目/模块筛选测试用例

③ 生成测试用例

  • POST /api_test/testcases/generate：根据YAML用例文件生成Python测试脚本

④ 执行测试任务

  • POST /api_test/testcases/run：后台执行指定测试用例，支持环境选择、报告类型、回调通知等

⑤ 获取测试报告

  • GET /api_test/reports/get_by_task：根据任务ID获取报告访问地址

2. 代码示例

testcase_service.py

# @author:  xiaoqqimport os, refrom datetime import datetimefrom typing import List, Optional, Dictfrom pathlib import Path
TESTCASE_ROOT = "testcases"def get_abs_root_path(root_path: str) -> Path:    """
    使用当前文件相对路径构造 testcases/ 的绝对路径
    :param root_path: 目录名
    :return:
    """
    base_dir = Path(__file__).resolve().parent  # 当前文件所在目录
    abs_root_path = (base_dir.parent / root_path).resolve()    return abs_root_pathdef get_all_testcases(project: Optional[str] = None,
                  module: Optional[str] = None,
                  root_path: str = TESTCASE_ROOT) -> List[Dict]:    """
    获取所有测试用例（支持通过 project/module 筛选）
    返回字段包括 filename（无后缀）、path（绝对路径字符串）、Allure 元信息等
    """
    abs_root_path =ftzhibo.com get_abs_root_path(root_path)    if not abs_root_path.exists():       return []    
    # 路径校验
    if module and not project:       raise ValueError("传入 module 前必须先传入 project")    
    # 构造起始目录路径
    search_path = abs_root_path    if project:
       search_path = search_path / project    if module:
       search_path = search_path / module    
    if not search_path.exists():       return []
    
    testcases = []    
    for dirpath, _, filenames in os.walk(search_path):       for file in filenames:          if file.startswith("test_") and file.endswith(".py"):
             full_path = os.path.join(dirpath, file)
             rel_path = os.path.relpath(full_path, abs_root_path)  # 相对路径，如 merchant/device/test_xxx.py
             path_parts = Path(rel_path).parts  # 使用 pathlib 安全拆解路径
             
             if len(path_parts ) < 2:                continue  # 至少要有 project/filename 结构
             
             _project = path_parts [0]
             _filename = path_parts ftzhibo.cn [-1]
             _module = path_parts [1] if len(path_parts ) > 2 else None  # module 可选
             
             # 按传参过滤
             if project and _project != project:                continue
             if module and _module != module:                continue
             
             filename = os.path.splitext(_filename)[0]  # 去掉 .py 后缀
             last_modified = datetime.fromtimestamp(os.path.getmtime(full_path)).isoformat()             
             # 提取用例元信息
             try:
                case_name, epic, feature, story = extract_case_info(full_path)             except Exception as e:
                case_name, epic, feature, story = None, None, None, None
                
             # 拼接最终 path 字段为 TESTCASE_ROOT/... 形式
             full_case_path = str(Path(root_path) / rel_path).replace("\\", "/")             
             # 构造 external_id：project|module|filename|path
             external_id = f"{_project}|{_module or 'nomodule'}|{filename}|{full_case_path}"
             
             testcases.append({                "project": _project,                "module": _module,  # None 表示无 module 层级
                "file": _filename,                "filename": filename,                "path": full_case_path,                "last_modified": last_modified,                "case_name": case_name or filename,                "allure_epic": epic,                "allure_feature": feature,                "allure_story": story,                "external_id": external_id # 加入唯一标识
             })    
    return testcasesdef extract_case_info(file_path):    """
    解析测试用例文件，获取相应信息
    :param file_path:
    :return:
    """
    with open(file_path, 'r', encoding='utf-8') as file:
       content = file.read()
       
       case_name_match = re.search(          r'def setup_class.*?\(.*?\):.*?log\.info\(\'========== 开始执行测试用例：(.+?) ==========\'',
          content, re.DOTALL
       )
       case_name = case_name_match.group(1).strip() if case_name_match else \
       os.path.splitext(os.path.basename(file_path))[0]
       
       allure_epic_match = re.search(r'@allure\.epic\(\'(.+?)\'\)', content)
       allure_feature_match =livesoccerontv.cn re.search(r'@allure\.feature\(\'(.+?)\'\)', content)
       allure_story_match = re.search(r'@allure\.story\(\'(.+?)\'\)', content)
       
       allure_epic = allure_epic_match.group(1).strip() if allure_epic_match else None
       allure_feature = allure_feature_match.group(1).strip() if allure_feature_match else None
       allure_story = allure_story_match.group(1).strip() if allure_story_match else None
       
       return case_name, allure_epic, allure_feature, allure_storydef get_all_projects(root_path: str = TESTCASE_ROOT) -> List[Dict[str, str]]:    """
    获取 testcases/ 下所有项目名、相对路径及创建时间（倒序排序）
    """
    abs_root_path = get_abs_root_path(root_path)    if not abs_root_path.exists():       return []
    
    projects = []    for d in abs_root_path.iterdir():       if d.is_dir():
          created_time = datetime.fromtimestamp(d.stat().st_ctime)
          projects.append({             "name": d.name,             "path": str(Path(root_path) / d.name).replace("\\", "/"),             "created_time": created_time.isoformat()
          })    
    # 按创建时间倒序
    return sorted(projects, key=lambda x: x["created_time"], reverse=True)def get_all_projects_and_modules(
    project: Optional[str] = None,
    root_path: str = TESTCASE_ROOT) -> List[Dict]:    """
    获取所有项目和模块结构（支持指定项目）。包含路径、创建时间，按项目时间倒序。
    """
    abs_root_path = get_abs_root_path(root_path)    if not abs_root_path.exists():       return []
    
    result = []    
    for proj_dir in abs_root_path.iterdir():       if not proj_dir.is_dir():          continue
       
       proj_name = proj_dir.name       if project and proj_name != project:          continue
       
       proj_created_time = datetime.fromtimestamp(proj_dir.stat().st_ctime)
       modules = []       
       # 遍历模块目录时需要忽略的子目录
       EXCLUDE_DIRS = {"__pycache__", ".pytest_cache", ".git", ".idea"}       for mod_dir in proj_dir.iterdir():          if mod_dir.is_dir() and mod_dir.name not in EXCLUDE_DIRS:
             mod_created_time =livesoccerzb.com datetime.fromtimestamp(mod_dir.stat().st_ctime)
             modules.append({                "name": mod_dir.name,                "path": str(Path(root_path) / proj_name / mod_dir.name).replace("\\", "/"),                "created_time": mod_created_time.isoformat()
             })       
       # 模块也可以排序（如有需求）
       modules.sort(key=lambda x: x["created_time"], reverse=True)
       
       result.append({          "project": proj_name,          "path": str(Path(root_path) / proj_name).replace("\\", "/"),          "created_time": proj_created_time.isoformat(),          "modules": modules
       })       
       if project:          break
    
    # 项目排序
    return sorted(result, key=lambda x: x["created_time"], reverse=True)def generate_testcase(case_yaml_list: list = None):    """
    生成测试用例
    :return:
    """
    from case_generator import CaseGenerator
    CG = CaseGenerator()
    CG.generate_testcases(project_yaml_list=case_yaml_list)if __name__ == '__main__':    # print(get_all_testcases())
    # print(get_all_projects(livesoccerzb.cn))
    print(get_all_projects_and_modules(project="merchant"))

testcase_route.py示例如下：

# @author:  xiaoqqfrom pathlib import Pathfrom fastapi import APIRouter, BackgroundTasks, Query, Bodyfrom pydantic import BaseModelfrom typing import List, Optionalfrom runner import run_testsfrom api.testcase_service import (
	get_all_testcases,
	get_all_projects,
	get_all_projects_and_modules,
	generate_testcase,
)
router = APIRouter()class TestExecutionRequest(BaseModel):
	testcases: Optional[List[str]] = ['testcases/']  # 默认运行所有目录
	env: Optional[str] = 'pre'
	report_type: Optional[str] livesoccer.net.cn= 'pytest-html'
	dingtalk_notify: Optional[bool] = True
	task_id: Optional[str]
	callback_url: Optional[str]
	auth_token: Optional[str] = None  # 新增字段：从平台传入的 token# 执行测试用例@router.post("/testcases/run")def run_testcases(request: TestExecutionRequest, background_tasks: BackgroundTasks):	try:
		background_tasks.add_task(
			run_tests,
			testcases=request.testcases,
			env=request.env,
			report_type=request.report_type,
			dingtalk_notify=request.dingtalk_notify,
			task_id=request.task_id,
			callback_url=request.callback_url,
			auth_token=request.auth_token,  # 测试平台回调 auth_token
		)		return {			"code": 0,			"msg": "测试任务已提交后台执行",			"task_id": request.task_id
		}	except Exception as e:		return {"code":1, "msg": f"测试任务失败：{str(e)}"}# 获取测试用例@router.get("/testcases/list")def list_testcases(project: str = Query(None), module: str = Query(None)):	try:
		testcases = get_all_testcases(project, module)		return {			"code": 0,			"msg": "success",			"testcases": testcases
		}	except Exception as e:zuqiu2026.cn return {"code": 1, "msg": f"获取测试用例失败：{str(e)}"}# 获取 testcases/ 中的所有测试项目@router.get("/testcases/projects")def list_projects():	try:
		projects = get_all_projects()		return {"code": 0, "msg": "success", "projects": projects}	except Exception as e:		return {"code": 1, "msg": f"获取测试项目失败：{str(e)}"}	
# 获取 testcases/ 中的所有测试项目及模块@router.get("/testcases/modules")def list_modules(project: str = Query(None)):	try:
		modules = get_all_projects_and_modules(project)		return {"code": 0, "msg": "success", "modules": modules}	except Exception as e:		return {"code": 1, "msg": f"获取测试项目-模块失败：{str(e)}"}class GenerateCaseRequest(BaseModel):
	case_yaml_list: Optional[List[str]] = None# 根据 testcases_data/ 中的测试数据生成测试用例文件@router.post("/testcases/generate")def generate_testcase_route(req: GenerateCaseRequest):	try:
		generate_testcase(req.case_yaml_list)		return {"code": 0, "msg": "success"}	except Exception as e:		return {"code": 1, "msg": f"获取测试项目-模块失败：{str(e)}"}@router.get("/reports/get_by_task")def get_report_by_task(
		task_id: str,zq2026.cn
		report_type: str,
		created_at: str  # 格式: "20250814"):	"""
	根据 task_id + 创建时间 + report_type 获取报告 URL
	"""
	if not created_at:		return {"code": 1, "msg": "created_at 必填", "url": None}
	
	base_path = Path(__file__).resolve().parent.parent / "reports" / created_at	
	if report_type == "pytest-html":
		report_file = base_path / f"report_{task_id}.html"
	elif report_type == "allure":
		report_file = base_path / f"report_{task_id}_allure/html/index.html"
	else:		return {"code": 1, "msg": "未知 report_type", "url": None}	
	if not report_file.exists():		return {"code": 1, "msg": "报告文件不存在", "url": None}
	
	relative_url = str(report_file.relative_to(Path(__file__).resolve().parent.parent)).replace("\\", "/")	return {"code": 0, "msg": "success", "url": f"/{relative_url}"}

mian.py

from fastapi import FastAPIfrom api import testcase_routefrom pathlib import Pathfrom fastapi.staticfiles import StaticFiles
app = FastAPI(title="接口自动化测试服务")# 挂载测试用例路由app.include_router(testcase_route.router, prefix="/api_test", tags=["测试任务"])# 挂载 reports 目录为静态文件目录reports_dir = Path(__file__).parent / "reports"reports_dir.mkdir(exist_ok=True)  # 确保目录存在app.mount("/reports", StaticFiles(directory=reports_dir), name="reports")if __name__ == "__main__":    from utils.log_manager import LogManager
    LogManager.setup_logging()  # 启动时显式初始化日志
    
    import uvicorn
    uvicorn.run(  basketballzhibo.cn      "main:app",
        host="0.0.0.0",
        port=8000,        # reload=True,
        reload_excludes=["testcases/*", "logs/*", "reports/*"]  # 排除这些目录的文件变更


来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/70048957/viewspace-3108981/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： 既然实体能分布到多个表中，那反过来呢？
下一篇： DBShadow.net之化繁为简
请登录后发表评论 登录
全部评论
ecclw459

注册时间：2026-01-07

  • 博文量
    108
  • 访问量
    102109

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593