refactor: enforce 800 line source limits
This commit is contained in:
caoxiaozhu
@@ -46,17 +46,305 @@ from app.services.risk_rule_score_backfill import backfill_missing_risk_rule_sco
|
||||
logger = get_logger("app.services.agent_assets")
|
||||
|
||||
|
||||
class AgentAssetService(
|
||||
AgentAssetOnlyOfficeMixin,
|
||||
AgentAssetSpreadsheetHelperMixin,
|
||||
AgentAssetRiskRuleLevelMixin,
|
||||
AgentAssetRiskRulePublishMixin,
|
||||
AgentAssetRiskRuleFeedbackMixin,
|
||||
AgentAssetRiskRuleTestingMixin,
|
||||
AgentAssetRiskRuleSimulationMixin,
|
||||
AgentAssetTimelineMixin,
|
||||
AgentAssetJsonRuleMixin,
|
||||
):
|
||||
class AgentAssetVersionMixin:
|
||||
def _validate_version_payload(
|
||||
self, asset: AgentAsset, payload: AgentAssetVersionCreate
|
||||
) -> None:
|
||||
if (
|
||||
asset.asset_type == AgentAssetType.RULE.value
|
||||
and payload.content_type != AgentAssetContentType.MARKDOWN
|
||||
):
|
||||
raise ValueError("规则资产版本内容必须使用 markdown。")
|
||||
if (
|
||||
asset.asset_type not in {AgentAssetType.RULE.value, AgentAssetType.TASK.value}
|
||||
and payload.content_type != AgentAssetContentType.JSON
|
||||
):
|
||||
raise ValueError("技能、MCP 资产版本内容必须使用 json。")
|
||||
if payload.content_type == AgentAssetContentType.MARKDOWN and not isinstance(
|
||||
payload.content, str
|
||||
):
|
||||
raise ValueError("Markdown 内容必须是字符串。")
|
||||
if payload.content_type == AgentAssetContentType.JSON and not isinstance(
|
||||
payload.content, (dict, list)
|
||||
):
|
||||
raise ValueError("JSON 内容必须是对象或数组。")
|
||||
|
||||
def restore_version_as_working_copy(
|
||||
self,
|
||||
asset_id: str,
|
||||
source_version: str,
|
||||
*,
|
||||
actor: str,
|
||||
request_id: str | None = None,
|
||||
) -> AgentAssetRead:
|
||||
self._ensure_ready()
|
||||
asset = self.repository.get(asset_id)
|
||||
if asset is None:
|
||||
raise LookupError("Asset not found")
|
||||
|
||||
source = self.repository.get_version(asset_id, source_version)
|
||||
if source is None:
|
||||
raise LookupError(f"版本 {source_version} 不存在")
|
||||
|
||||
if (
|
||||
asset.asset_type == AgentAssetType.RULE.value
|
||||
and str((asset.config_json or {}).get("detail_mode") or "").strip().lower()
|
||||
== "spreadsheet"
|
||||
):
|
||||
metadata = self.spreadsheet_manager.parse_version_markdown(str(source.content or ""))
|
||||
if metadata is None:
|
||||
raise FileNotFoundError("历史规则表快照不存在，无法恢复。")
|
||||
file_path = self.spreadsheet_manager.resolve_storage_path(metadata.storage_key)
|
||||
if not file_path.exists():
|
||||
raise FileNotFoundError(metadata.file_name)
|
||||
restored = self.upload_rule_spreadsheet(
|
||||
asset.id,
|
||||
filename=metadata.file_name,
|
||||
content=file_path.read_bytes(),
|
||||
actor=actor,
|
||||
request_id=request_id,
|
||||
change_note=f"基于历史版本 {source_version} 恢复生成工作稿",
|
||||
source="restore",
|
||||
)
|
||||
self.audit_service.log_action(
|
||||
actor=actor,
|
||||
action="restore_agent_asset_version",
|
||||
resource_type=asset.asset_type,
|
||||
resource_id=asset.id,
|
||||
before_json={"source_version": source_version},
|
||||
after_json={"working_version": restored.working_version},
|
||||
request_id=request_id,
|
||||
)
|
||||
return restored
|
||||
|
||||
next_version = self._increment_version(self._resolve_working_version(asset))
|
||||
self.create_version(
|
||||
asset.id,
|
||||
AgentAssetVersionCreate(
|
||||
version=next_version,
|
||||
content=self._deserialize_content(source),
|
||||
content_type=AgentAssetContentType(source.content_type),
|
||||
change_note=f"基于历史版本 {source_version} 恢复生成工作稿",
|
||||
created_by=actor,
|
||||
),
|
||||
actor=actor,
|
||||
request_id=request_id,
|
||||
)
|
||||
restored = self.get_asset(asset.id)
|
||||
self.audit_service.log_action(
|
||||
actor=actor,
|
||||
action="restore_agent_asset_version",
|
||||
resource_type=asset.asset_type,
|
||||
resource_id=asset.id,
|
||||
before_json={"source_version": source_version},
|
||||
after_json={"working_version": next_version},
|
||||
request_id=request_id,
|
||||
)
|
||||
return restored # type: ignore[return-value]
|
||||
|
||||
def _serialize_version(
|
||||
self, version: AgentAssetVersion, asset: AgentAsset
|
||||
) -> AgentAssetVersionRead:
|
||||
latest_review = self.repository.get_review(asset.id, version.version)
|
||||
working_version = self._resolve_working_version(asset)
|
||||
published_version = self._resolve_published_version(asset)
|
||||
return AgentAssetVersionRead(
|
||||
id=version.id,
|
||||
asset_id=version.asset_id,
|
||||
version=version.version,
|
||||
content=self._deserialize_content(version),
|
||||
content_type=version.content_type,
|
||||
change_note=version.change_note,
|
||||
created_by=version.created_by,
|
||||
created_at=version.created_at,
|
||||
is_current=version.version == working_version,
|
||||
is_published=version.version == published_version,
|
||||
is_working=version.version == working_version,
|
||||
lifecycle_state=self._resolve_version_lifecycle_state(
|
||||
version.version,
|
||||
working_version=working_version,
|
||||
published_version=published_version,
|
||||
latest_review_status=latest_review.review_status if latest_review else "",
|
||||
),
|
||||
)
|
||||
|
||||
def _collect_version_stats(self, assets: list[AgentAsset]) -> dict[str, dict[str, Any]]:
|
||||
asset_ids = [item.id for item in assets]
|
||||
versions = self.repository.list_versions_for_assets(asset_ids)
|
||||
reviews = self.repository.list_reviews_for_assets(asset_ids)
|
||||
spreadsheet_logs = self.audit_service.repository.list_for_resources(
|
||||
resource_type=AgentAssetType.RULE.value,
|
||||
resource_ids=[
|
||||
item.id
|
||||
for item in assets
|
||||
if item.asset_type == AgentAssetType.RULE.value
|
||||
and str((item.config_json or {}).get("detail_mode") or "").strip().lower()
|
||||
== "spreadsheet"
|
||||
],
|
||||
action="edit_rule_spreadsheet",
|
||||
)
|
||||
working_versions = {item.id: self._resolve_working_version(item) for item in assets}
|
||||
version_counts: dict[str, int] = defaultdict(int)
|
||||
modified_by: dict[str, str | None] = {item.id: None for item in assets}
|
||||
published_versions = {item.id: self._resolve_published_version(item) for item in assets}
|
||||
published_by: dict[str, str | None] = {}
|
||||
published_at: dict[str, datetime | None] = {}
|
||||
spreadsheet_edit_counts: dict[str, int] = defaultdict(int)
|
||||
spreadsheet_last_actor: dict[str, str | None] = {}
|
||||
spreadsheet_last_changed_at: dict[str, datetime] = {}
|
||||
|
||||
for version in versions:
|
||||
version_counts[version.asset_id] += 1
|
||||
if modified_by.get(
|
||||
version.asset_id
|
||||
) is None and version.version == working_versions.get(version.asset_id):
|
||||
modified_by[version.asset_id] = version.created_by
|
||||
|
||||
for review in reviews:
|
||||
if review.asset_id in published_at:
|
||||
continue
|
||||
if review.version != published_versions.get(review.asset_id):
|
||||
continue
|
||||
if review.review_status != AgentReviewStatus.APPROVED.value:
|
||||
continue
|
||||
published_by[review.asset_id] = review.reviewer
|
||||
published_at[review.asset_id] = review.reviewed_at or review.created_at
|
||||
|
||||
for log in spreadsheet_logs:
|
||||
spreadsheet_edit_counts[log.resource_id] += 1
|
||||
last_changed_at = spreadsheet_last_changed_at.get(log.resource_id)
|
||||
if last_changed_at is None or log.created_at >= last_changed_at:
|
||||
spreadsheet_last_changed_at[log.resource_id] = log.created_at
|
||||
spreadsheet_last_actor[log.resource_id] = log.actor
|
||||
|
||||
return {
|
||||
item.id: {
|
||||
"change_count": (
|
||||
spreadsheet_edit_counts.get(item.id, 0)
|
||||
if item.asset_type == AgentAssetType.RULE.value
|
||||
and str((item.config_json or {}).get("detail_mode") or "").strip().lower()
|
||||
== "spreadsheet"
|
||||
and spreadsheet_edit_counts.get(item.id, 0) > 0
|
||||
else max(version_counts.get(item.id, 0) - 1, 0)
|
||||
),
|
||||
"modified_by": (
|
||||
spreadsheet_last_actor.get(item.id)
|
||||
if item.asset_type == AgentAssetType.RULE.value
|
||||
and str((item.config_json or {}).get("detail_mode") or "").strip().lower()
|
||||
== "spreadsheet"
|
||||
and spreadsheet_last_actor.get(item.id)
|
||||
else modified_by.get(item.id)
|
||||
),
|
||||
"published_by": published_by.get(item.id),
|
||||
"published_at": published_at.get(item.id),
|
||||
}
|
||||
for item in assets
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _serialize_list_item(
|
||||
asset: AgentAsset,
|
||||
version_stats: dict[str, int | str | None] | None = None,
|
||||
) -> AgentAssetListItem:
|
||||
payload = AgentAssetListItem.model_validate(asset).model_dump()
|
||||
payload["change_count"] = int((version_stats or {}).get("change_count") or 0)
|
||||
payload["modified_by"] = str((version_stats or {}).get("modified_by") or "").strip() or None
|
||||
payload["published_by"] = (
|
||||
str((version_stats or {}).get("published_by") or "").strip() or None
|
||||
)
|
||||
payload["published_at"] = (version_stats or {}).get("published_at")
|
||||
return AgentAssetListItem.model_validate(payload)
|
||||
|
||||
@staticmethod
|
||||
def _sort_versions(
|
||||
versions: list[AgentAssetVersion], current_version: str | None
|
||||
) -> list[AgentAssetVersion]:
|
||||
return sorted(
|
||||
versions,
|
||||
key=lambda item: (item.version == current_version, item.created_at),
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _serialize_content(content: Any, content_type: str) -> str:
|
||||
if content_type == AgentAssetContentType.MARKDOWN.value:
|
||||
return str(content)
|
||||
return json.dumps(content, ensure_ascii=False, sort_keys=True, indent=2)
|
||||
|
||||
@staticmethod
|
||||
def _deserialize_content(version: AgentAssetVersion | None) -> Any:
|
||||
if version is None:
|
||||
return None
|
||||
if version.content_type == AgentAssetContentType.MARKDOWN.value:
|
||||
return version.content
|
||||
return json.loads(version.content)
|
||||
|
||||
@staticmethod
|
||||
def _increment_version(version: str | None) -> str:
|
||||
normalized = str(version or "").strip().removeprefix("v")
|
||||
parts = normalized.split(".")
|
||||
if len(parts) != 3 or not all(item.isdigit() for item in parts):
|
||||
return "v1.0.0"
|
||||
major, minor, patch = [int(item) for item in parts]
|
||||
return f"v{major}.{minor}.{patch + 1}"
|
||||
|
||||
@staticmethod
|
||||
def _hash_bytes(content: bytes) -> str:
|
||||
import hashlib
|
||||
|
||||
return hashlib.sha256(content).hexdigest()
|
||||
|
||||
@staticmethod
|
||||
def _asset_snapshot(asset: AgentAsset) -> dict[str, Any]:
|
||||
return {
|
||||
"asset_type": asset.asset_type,
|
||||
"code": asset.code,
|
||||
"name": asset.name,
|
||||
"status": asset.status,
|
||||
"current_version": asset.current_version,
|
||||
"published_version": asset.published_version,
|
||||
"working_version": asset.working_version,
|
||||
"domain": asset.domain,
|
||||
"owner": asset.owner,
|
||||
"reviewer": asset.reviewer,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _resolve_working_version(asset: AgentAsset) -> str:
|
||||
return str(asset.working_version or asset.current_version or "").strip()
|
||||
|
||||
@staticmethod
|
||||
def _resolve_published_version(asset: AgentAsset) -> str:
|
||||
return str(asset.published_version or "").strip()
|
||||
|
||||
@staticmethod
|
||||
def _resolve_version_lifecycle_state(
|
||||
version: str,
|
||||
*,
|
||||
working_version: str,
|
||||
published_version: str,
|
||||
latest_review_status: str,
|
||||
) -> str:
|
||||
if version == published_version:
|
||||
return "published"
|
||||
if version != working_version:
|
||||
return "history"
|
||||
if latest_review_status == AgentReviewStatus.PENDING.value:
|
||||
return "pending_review"
|
||||
if latest_review_status == AgentReviewStatus.APPROVED.value:
|
||||
return "approved"
|
||||
if latest_review_status == AgentReviewStatus.REJECTED.value:
|
||||
return "rejected"
|
||||
return "draft"
|
||||
|
||||
def _next_available_version(self, asset: AgentAsset) -> str:
|
||||
candidate = self._increment_version(self._resolve_working_version(asset))
|
||||
while self.repository.get_version(asset.id, candidate) is not None:
|
||||
candidate = self._increment_version(candidate)
|
||||
return candidate
|
||||
|
||||
|
||||
class AgentAssetService(AgentAssetVersionMixin, AgentAssetOnlyOfficeMixin, AgentAssetSpreadsheetHelperMixin, AgentAssetRiskRuleLevelMixin, AgentAssetRiskRulePublishMixin, AgentAssetRiskRuleFeedbackMixin, AgentAssetRiskRuleTestingMixin, AgentAssetRiskRuleSimulationMixin, AgentAssetTimelineMixin, AgentAssetJsonRuleMixin):
|
||||
def __init__(self, db: Session) -> None:
|
||||
self.db = db
|
||||
self.repository = AgentAssetRepository(db)
|
||||
@@ -559,298 +847,3 @@ class AgentAssetService(
|
||||
self.db.commit()
|
||||
return synced_count
|
||||
|
||||
def _validate_version_payload(
|
||||
self, asset: AgentAsset, payload: AgentAssetVersionCreate
|
||||
) -> None:
|
||||
if (
|
||||
asset.asset_type == AgentAssetType.RULE.value
|
||||
and payload.content_type != AgentAssetContentType.MARKDOWN
|
||||
):
|
||||
raise ValueError("规则资产版本内容必须使用 markdown。")
|
||||
if (
|
||||
asset.asset_type not in {AgentAssetType.RULE.value, AgentAssetType.TASK.value}
|
||||
and payload.content_type != AgentAssetContentType.JSON
|
||||
):
|
||||
raise ValueError("技能、MCP 资产版本内容必须使用 json。")
|
||||
if payload.content_type == AgentAssetContentType.MARKDOWN and not isinstance(
|
||||
payload.content, str
|
||||
):
|
||||
raise ValueError("Markdown 内容必须是字符串。")
|
||||
if payload.content_type == AgentAssetContentType.JSON and not isinstance(
|
||||
payload.content, (dict, list)
|
||||
):
|
||||
raise ValueError("JSON 内容必须是对象或数组。")
|
||||
|
||||
def restore_version_as_working_copy(
|
||||
self,
|
||||
asset_id: str,
|
||||
source_version: str,
|
||||
*,
|
||||
actor: str,
|
||||
request_id: str | None = None,
|
||||
) -> AgentAssetRead:
|
||||
self._ensure_ready()
|
||||
asset = self.repository.get(asset_id)
|
||||
if asset is None:
|
||||
raise LookupError("Asset not found")
|
||||
|
||||
source = self.repository.get_version(asset_id, source_version)
|
||||
if source is None:
|
||||
raise LookupError(f"版本 {source_version} 不存在")
|
||||
|
||||
if (
|
||||
asset.asset_type == AgentAssetType.RULE.value
|
||||
and str((asset.config_json or {}).get("detail_mode") or "").strip().lower()
|
||||
== "spreadsheet"
|
||||
):
|
||||
metadata = self.spreadsheet_manager.parse_version_markdown(str(source.content or ""))
|
||||
if metadata is None:
|
||||
raise FileNotFoundError("历史规则表快照不存在，无法恢复。")
|
||||
file_path = self.spreadsheet_manager.resolve_storage_path(metadata.storage_key)
|
||||
if not file_path.exists():
|
||||
raise FileNotFoundError(metadata.file_name)
|
||||
restored = self.upload_rule_spreadsheet(
|
||||
asset.id,
|
||||
filename=metadata.file_name,
|
||||
content=file_path.read_bytes(),
|
||||
actor=actor,
|
||||
request_id=request_id,
|
||||
change_note=f"基于历史版本 {source_version} 恢复生成工作稿",
|
||||
source="restore",
|
||||
)
|
||||
self.audit_service.log_action(
|
||||
actor=actor,
|
||||
action="restore_agent_asset_version",
|
||||
resource_type=asset.asset_type,
|
||||
resource_id=asset.id,
|
||||
before_json={"source_version": source_version},
|
||||
after_json={"working_version": restored.working_version},
|
||||
request_id=request_id,
|
||||
)
|
||||
return restored
|
||||
|
||||
next_version = self._increment_version(self._resolve_working_version(asset))
|
||||
self.create_version(
|
||||
asset.id,
|
||||
AgentAssetVersionCreate(
|
||||
version=next_version,
|
||||
content=self._deserialize_content(source),
|
||||
content_type=AgentAssetContentType(source.content_type),
|
||||
change_note=f"基于历史版本 {source_version} 恢复生成工作稿",
|
||||
created_by=actor,
|
||||
),
|
||||
actor=actor,
|
||||
request_id=request_id,
|
||||
)
|
||||
restored = self.get_asset(asset.id)
|
||||
self.audit_service.log_action(
|
||||
actor=actor,
|
||||
action="restore_agent_asset_version",
|
||||
resource_type=asset.asset_type,
|
||||
resource_id=asset.id,
|
||||
before_json={"source_version": source_version},
|
||||
after_json={"working_version": next_version},
|
||||
request_id=request_id,
|
||||
)
|
||||
return restored # type: ignore[return-value]
|
||||
|
||||
def _serialize_version(
|
||||
self, version: AgentAssetVersion, asset: AgentAsset
|
||||
) -> AgentAssetVersionRead:
|
||||
latest_review = self.repository.get_review(asset.id, version.version)
|
||||
working_version = self._resolve_working_version(asset)
|
||||
published_version = self._resolve_published_version(asset)
|
||||
return AgentAssetVersionRead(
|
||||
id=version.id,
|
||||
asset_id=version.asset_id,
|
||||
version=version.version,
|
||||
content=self._deserialize_content(version),
|
||||
content_type=version.content_type,
|
||||
change_note=version.change_note,
|
||||
created_by=version.created_by,
|
||||
created_at=version.created_at,
|
||||
is_current=version.version == working_version,
|
||||
is_published=version.version == published_version,
|
||||
is_working=version.version == working_version,
|
||||
lifecycle_state=self._resolve_version_lifecycle_state(
|
||||
version.version,
|
||||
working_version=working_version,
|
||||
published_version=published_version,
|
||||
latest_review_status=latest_review.review_status if latest_review else "",
|
||||
),
|
||||
)
|
||||
|
||||
def _collect_version_stats(self, assets: list[AgentAsset]) -> dict[str, dict[str, Any]]:
|
||||
asset_ids = [item.id for item in assets]
|
||||
versions = self.repository.list_versions_for_assets(asset_ids)
|
||||
reviews = self.repository.list_reviews_for_assets(asset_ids)
|
||||
spreadsheet_logs = self.audit_service.repository.list_for_resources(
|
||||
resource_type=AgentAssetType.RULE.value,
|
||||
resource_ids=[
|
||||
item.id
|
||||
for item in assets
|
||||
if item.asset_type == AgentAssetType.RULE.value
|
||||
and str((item.config_json or {}).get("detail_mode") or "").strip().lower()
|
||||
== "spreadsheet"
|
||||
],
|
||||
action="edit_rule_spreadsheet",
|
||||
)
|
||||
working_versions = {item.id: self._resolve_working_version(item) for item in assets}
|
||||
version_counts: dict[str, int] = defaultdict(int)
|
||||
modified_by: dict[str, str | None] = {item.id: None for item in assets}
|
||||
published_versions = {item.id: self._resolve_published_version(item) for item in assets}
|
||||
published_by: dict[str, str | None] = {}
|
||||
published_at: dict[str, datetime | None] = {}
|
||||
spreadsheet_edit_counts: dict[str, int] = defaultdict(int)
|
||||
spreadsheet_last_actor: dict[str, str | None] = {}
|
||||
spreadsheet_last_changed_at: dict[str, datetime] = {}
|
||||
|
||||
for version in versions:
|
||||
version_counts[version.asset_id] += 1
|
||||
if modified_by.get(
|
||||
version.asset_id
|
||||
) is None and version.version == working_versions.get(version.asset_id):
|
||||
modified_by[version.asset_id] = version.created_by
|
||||
|
||||
for review in reviews:
|
||||
if review.asset_id in published_at:
|
||||
continue
|
||||
if review.version != published_versions.get(review.asset_id):
|
||||
continue
|
||||
if review.review_status != AgentReviewStatus.APPROVED.value:
|
||||
continue
|
||||
published_by[review.asset_id] = review.reviewer
|
||||
published_at[review.asset_id] = review.reviewed_at or review.created_at
|
||||
|
||||
for log in spreadsheet_logs:
|
||||
spreadsheet_edit_counts[log.resource_id] += 1
|
||||
last_changed_at = spreadsheet_last_changed_at.get(log.resource_id)
|
||||
if last_changed_at is None or log.created_at >= last_changed_at:
|
||||
spreadsheet_last_changed_at[log.resource_id] = log.created_at
|
||||
spreadsheet_last_actor[log.resource_id] = log.actor
|
||||
|
||||
return {
|
||||
item.id: {
|
||||
"change_count": (
|
||||
spreadsheet_edit_counts.get(item.id, 0)
|
||||
if item.asset_type == AgentAssetType.RULE.value
|
||||
and str((item.config_json or {}).get("detail_mode") or "").strip().lower()
|
||||
== "spreadsheet"
|
||||
and spreadsheet_edit_counts.get(item.id, 0) > 0
|
||||
else max(version_counts.get(item.id, 0) - 1, 0)
|
||||
),
|
||||
"modified_by": (
|
||||
spreadsheet_last_actor.get(item.id)
|
||||
if item.asset_type == AgentAssetType.RULE.value
|
||||
and str((item.config_json or {}).get("detail_mode") or "").strip().lower()
|
||||
== "spreadsheet"
|
||||
and spreadsheet_last_actor.get(item.id)
|
||||
else modified_by.get(item.id)
|
||||
),
|
||||
"published_by": published_by.get(item.id),
|
||||
"published_at": published_at.get(item.id),
|
||||
}
|
||||
for item in assets
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _serialize_list_item(
|
||||
asset: AgentAsset,
|
||||
version_stats: dict[str, int | str | None] | None = None,
|
||||
) -> AgentAssetListItem:
|
||||
payload = AgentAssetListItem.model_validate(asset).model_dump()
|
||||
payload["change_count"] = int((version_stats or {}).get("change_count") or 0)
|
||||
payload["modified_by"] = str((version_stats or {}).get("modified_by") or "").strip() or None
|
||||
payload["published_by"] = (
|
||||
str((version_stats or {}).get("published_by") or "").strip() or None
|
||||
)
|
||||
payload["published_at"] = (version_stats or {}).get("published_at")
|
||||
return AgentAssetListItem.model_validate(payload)
|
||||
|
||||
@staticmethod
|
||||
def _sort_versions(
|
||||
versions: list[AgentAssetVersion], current_version: str | None
|
||||
) -> list[AgentAssetVersion]:
|
||||
return sorted(
|
||||
versions,
|
||||
key=lambda item: (item.version == current_version, item.created_at),
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _serialize_content(content: Any, content_type: str) -> str:
|
||||
if content_type == AgentAssetContentType.MARKDOWN.value:
|
||||
return str(content)
|
||||
return json.dumps(content, ensure_ascii=False, sort_keys=True, indent=2)
|
||||
|
||||
@staticmethod
|
||||
def _deserialize_content(version: AgentAssetVersion | None) -> Any:
|
||||
if version is None:
|
||||
return None
|
||||
if version.content_type == AgentAssetContentType.MARKDOWN.value:
|
||||
return version.content
|
||||
return json.loads(version.content)
|
||||
|
||||
@staticmethod
|
||||
def _increment_version(version: str | None) -> str:
|
||||
normalized = str(version or "").strip().removeprefix("v")
|
||||
parts = normalized.split(".")
|
||||
if len(parts) != 3 or not all(item.isdigit() for item in parts):
|
||||
return "v1.0.0"
|
||||
major, minor, patch = [int(item) for item in parts]
|
||||
return f"v{major}.{minor}.{patch + 1}"
|
||||
|
||||
@staticmethod
|
||||
def _hash_bytes(content: bytes) -> str:
|
||||
import hashlib
|
||||
|
||||
return hashlib.sha256(content).hexdigest()
|
||||
|
||||
@staticmethod
|
||||
def _asset_snapshot(asset: AgentAsset) -> dict[str, Any]:
|
||||
return {
|
||||
"asset_type": asset.asset_type,
|
||||
"code": asset.code,
|
||||
"name": asset.name,
|
||||
"status": asset.status,
|
||||
"current_version": asset.current_version,
|
||||
"published_version": asset.published_version,
|
||||
"working_version": asset.working_version,
|
||||
"domain": asset.domain,
|
||||
"owner": asset.owner,
|
||||
"reviewer": asset.reviewer,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _resolve_working_version(asset: AgentAsset) -> str:
|
||||
return str(asset.working_version or asset.current_version or "").strip()
|
||||
|
||||
@staticmethod
|
||||
def _resolve_published_version(asset: AgentAsset) -> str:
|
||||
return str(asset.published_version or "").strip()
|
||||
|
||||
@staticmethod
|
||||
def _resolve_version_lifecycle_state(
|
||||
version: str,
|
||||
*,
|
||||
working_version: str,
|
||||
published_version: str,
|
||||
latest_review_status: str,
|
||||
) -> str:
|
||||
if version == published_version:
|
||||
return "published"
|
||||
if version != working_version:
|
||||
return "history"
|
||||
if latest_review_status == AgentReviewStatus.PENDING.value:
|
||||
return "pending_review"
|
||||
if latest_review_status == AgentReviewStatus.APPROVED.value:
|
||||
return "approved"
|
||||
if latest_review_status == AgentReviewStatus.REJECTED.value:
|
||||
return "rejected"
|
||||
return "draft"
|
||||
|
||||
def _next_available_version(self, asset: AgentAsset) -> str:
|
||||
candidate = self._increment_version(self._resolve_working_version(asset))
|
||||
while self.repository.get_version(asset.id, candidate) is not None:
|
||||
candidate = self._increment_version(candidate)
|
||||
return candidate
|
||||
|
||||
Reference in New Issue
Block a user