busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

log refactor

Browse Source
This commit is contained in:
busya
2026-05-12 19:30:15 +03:00
parent 1d59df2233
commit b17b5333c7
84 changed files with 5827 additions and 3908 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
backend/src/plugins/git_plugin.py
View File

@@ -22,6 +22,9 @@ from typing import Dict, Any, Optional
from src.core.plugin_base import PluginBase
from src.services.git_service import GitService
from src.core.logger import logger as app_logger, belief_scope
from src.core.cot_logger import MarkerLogger
log = MarkerLogger("GitPlugin")
from src.core.config_manager import ConfigManager
from src.core.superset_client import SupersetClient
from src.core.task_manager.context import TaskContext
@@ -39,7 +42,7 @@ class GitPlugin(PluginBase):
# @POST: Инициализированы git_service и config_manager.
def __init__(self):
with belief_scope("GitPlugin.__init__"):
app_logger.info("Initializing GitPlugin.")
log.reason("Initializing GitPlugin.")
self.git_service = GitService()
# Robust config path resolution:
@@ -54,13 +57,13 @@ class GitPlugin(PluginBase):
try:
from src.dependencies import config_manager
self.config_manager = config_manager
app_logger.info("GitPlugin initialized using shared config_manager.")
log.reflect("GitPlugin initialized using shared config_manager.")
return
except Exception:
config_path = "config.json"
self.config_manager = ConfigManager(config_path)
app_logger.info(f"GitPlugin initialized with {config_path}")
log.reflect(f"GitPlugin initialized with {config_path}")
# [/DEF:__init__:Function]
@property
@@ -137,7 +140,7 @@ class GitPlugin(PluginBase):
# @POST: Плагин готов к выполнению задач.
async def initialize(self):
with belief_scope("GitPlugin.initialize"):
app_logger.info("[GitPlugin.initialize][Action] Initializing Git Integration Plugin logic.")
log.reason("Initializing Git Integration Plugin logic")
# [DEF:execute:Function]
# @PURPOSE: Основной метод выполнения задач плагина с поддержкой TaskContext.
@@ -248,15 +251,15 @@ class GitPlugin(PluginBase):
# 5. Автоматический staging изменений (не коммит, чтобы юзер мог проверить diff)
try:
repo.git.add(A=True)
app_logger.info("[_handle_sync][Action] Changes staged in git")
log.reason("Changes staged in git")
except Exception as ge:
app_logger.warning(f"[_handle_sync][Action] Failed to stage changes: {ge}")
log.explore("Failed to stage changes", error=str(ge))
app_logger.info(f"[_handle_sync][Coherence:OK] Dashboard {dashboard_id} synced successfully.")
log.reflect("Dashboard synced successfully", payload={"dashboard_id": dashboard_id})
return {"status": "success", "message": "Dashboard synced and flattened in local repository"}
except Exception as e:
app_logger.error(f"[_handle_sync][Coherence:Failed] Sync failed: {e}")
log.explore("Sync failed", error=str(e))
raise
# [/DEF:_handle_sync:Function]
@@ -318,16 +321,16 @@ class GitPlugin(PluginBase):
f.write(zip_buffer.getvalue())
try:
app_logger.info(f"[_handle_deploy][Action] Importing dashboard to {env.name}")
log.reason("Importing dashboard", payload={"environment": env.name})
result = client.import_dashboard(temp_zip_path)
app_logger.info(f"[_handle_deploy][Coherence:OK] Deployment successful for dashboard {dashboard_id}.")
log.reflect("Deployment successful", payload={"dashboard_id": dashboard_id})
return {"status": "success", "message": f"Dashboard deployed to {env.name}", "details": result}
finally:
if temp_zip_path.exists():
os.remove(temp_zip_path)
except Exception as e:
app_logger.error(f"[_handle_deploy][Coherence:Failed] Deployment failed: {e}")
log.explore("Deployment failed", error=str(e))
raise
# [/DEF:_handle_deploy:Function]
@@ -339,13 +342,13 @@ class GitPlugin(PluginBase):
# @RETURN: Environment - Объект конфигурации окружения.
def _get_env(self, env_id: Optional[str] = None):
with belief_scope("GitPlugin._get_env"):
app_logger.info(f"[_get_env][Entry] Fetching environment for ID: {env_id}")
log.reason("Fetching environment", payload={"env_id": env_id})
# Priority 1: ConfigManager (config.json)
if env_id:
env = self.config_manager.get_environment(env_id)
if env:
app_logger.info(f"[_get_env][Exit] Found environment by ID in ConfigManager: {env.name}")
log.reflect("Found environment by ID in ConfigManager", payload={"name": env.name})
return env
# Priority 2: Database (DeploymentEnvironment)
@@ -363,7 +366,7 @@ class GitPlugin(PluginBase):
db_env = db.query(DeploymentEnvironment).first()
if db_env:
app_logger.info(f"[_get_env][Exit] Found environment in DB: {db_env.name}")
log.reflect("Found environment in DB", payload={"name": db_env.name})
from src.core.config_models import Environment
# Use token as password for SupersetClient
return Environment(
@@ -385,14 +388,14 @@ class GitPlugin(PluginBase):
# but we have other envs, maybe it's one of them?
env = next((e for e in envs if e.id == env_id), None)
if env:
app_logger.info(f"[_get_env][Exit] Found environment {env_id} in ConfigManager list")
log.reflect("Found environment in ConfigManager list", payload={"env_id": env_id})
return env
if not env_id:
app_logger.info(f"[_get_env][Exit] Using first environment from ConfigManager: {envs[0].name}")
log.reflect("Using first environment from ConfigManager", payload={"name": envs[0].name})
return envs[0]
app_logger.error(f"[_get_env][Coherence:Failed] No environments configured (searched config.json and DB). env_id={env_id}")
log.explore("No environments configured", payload={"env_id": env_id}, error="No Superset environments found in config or database")
raise ValueError("No environments configured. Please add a Superset Environment in Settings.")
# [/DEF:_get_env:Function]

61
backend/src/plugins/migration.py
View File

@@ -18,8 +18,11 @@ from typing import Dict, Any, Optional
import re
from ..core.plugin_base import PluginBase
from ..core.logger import belief_scope, logger as app_logger
from ..core.logger import belief_scope
from ..core.cot_logger import MarkerLogger
from ..core.superset_client import SupersetClient
log = MarkerLogger("MigrationPlugin")
from ..core.utils.fileio import create_temp_file
from ..dependencies import get_config_manager
from ..core.migration_engine import MigrationEngine
@@ -105,7 +108,7 @@ class MigrationPlugin(PluginBase):
# @RETURN: Dict[str, Any]
def get_schema(self) -> Dict[str, Any]:
with belief_scope("MigrationPlugin.get_schema"):
app_logger.reason("Generating migration UI schema")
log.reason("Generating migration UI schema")
config_manager = get_config_manager()
envs = [e.name for e in config_manager.get_environments()]
@@ -148,7 +151,7 @@ class MigrationPlugin(PluginBase):
},
"required": ["from_env", "to_env", "dashboard_regex"],
}
app_logger.reflect("Schema generated successfully", extra={"environments_count": len(envs)})
log.reflect("Schema generated successfully", payload={"environments_count": len(envs)})
return schema
# [/DEF:get_schema:Function]
@@ -168,7 +171,7 @@ class MigrationPlugin(PluginBase):
# @TEST_EDGE: target_api_timeout -> [Dashboard added to failed_dashboards, task concludes with PARTIAL_SUCCESS]
async def execute(self, params: Dict[str, Any], context: Optional[TaskContext] = None):
with belief_scope("MigrationPlugin.execute"):
app_logger.reason("Evaluating migration task parameters", extra={"params": params})
log.reason("Evaluating migration task parameters", payload={"params": params})
source_env_id = params.get("source_env_id")
target_env_id = params.get("target_env_id")
@@ -184,11 +187,11 @@ class MigrationPlugin(PluginBase):
from ..dependencies import get_task_manager
tm = get_task_manager()
log = context.logger if context else app_logger
superset_log = log.with_source("superset_api") if context else log
migration_log = log.with_source("migration") if context else log
exec_log = context.logger if context else log
superset_log = exec_log.with_source("superset_api") if context else exec_log
migration_log = exec_log.with_source("migration") if context else exec_log
log.info("Starting migration task.")
exec_log.info("Starting migration task.")
try:
config_manager = get_config_manager()
@@ -199,13 +202,13 @@ class MigrationPlugin(PluginBase):
tgt_env = next((e for e in environments if e.id == target_env_id), None) if target_env_id else next((e for e in environments if e.name == to_env_name), None)
if not src_env or not tgt_env:
app_logger.explore("Environment resolution failed", extra={"src": source_env_id or from_env_name, "tgt": target_env_id or to_env_name})
log.explore("Environment resolution failed", error="Could not resolve source or target environment", payload={"src": source_env_id or from_env_name, "tgt": target_env_id or to_env_name})
raise ValueError(f"Could not resolve source or target environment. Source: {source_env_id or from_env_name}, Target: {target_env_id or to_env_name}")
from_env_name = src_env.name
to_env_name = tgt_env.name
app_logger.reason("Environments resolved successfully", extra={"from": from_env_name, "to": to_env_name})
log.reason("Environments resolved successfully", payload={"from": from_env_name, "to": to_env_name})
migration_result = {
"status": "SUCCESS",
@@ -232,12 +235,12 @@ class MigrationPlugin(PluginBase):
regex_pattern = re.compile(str(dashboard_regex), re.IGNORECASE)
dashboards_to_migrate = [d for d in all_dashboards if regex_pattern.search(d.get("dashboard_title", ""))]
else:
app_logger.explore("No deterministic selection criteria provided")
log.explore("No deterministic selection criteria provided", error="No dashboard selection criteria (selected_ids or dashboard_regex)")
migration_result["status"] = "NO_SELECTION"
return migration_result
if not dashboards_to_migrate:
app_logger.explore("Zero dashboards match selection criteria")
log.explore("Zero dashboards match selection criteria", error="No dashboards matched the given selection criteria")
migration_result["status"] = "NO_MATCHES"
return migration_result
@@ -249,7 +252,7 @@ class MigrationPlugin(PluginBase):
db_mapping = {}
if replace_db_config:
app_logger.reason("Fetching environment DB mappings from catalog")
log.reason("Fetching environment DB mappings from catalog")
db = SessionLocal()
try:
src_env_db = db.query(Environment).filter(Environment.name == from_env_name).first()
@@ -263,7 +266,7 @@ class MigrationPlugin(PluginBase):
stored_map_dict = {m.source_db_uuid: m.target_db_uuid for m in stored_mappings}
stored_map_dict.update(db_mapping)
db_mapping = stored_map_dict
log.info(f"Loaded {len(stored_mappings)} database mappings from database.")
exec_log.info(f"Loaded {len(stored_mappings)} database mappings from database.")
finally:
db.close()
@@ -273,7 +276,7 @@ class MigrationPlugin(PluginBase):
# Migration Loop
for dash in dashboards_to_migrate:
dash_id, dash_slug, title = dash["id"], dash.get("slug"), dash["dashboard_title"]
app_logger.reason(f"Starting pipeline for dashboard '{title}'", extra={"dash_id": dash_id})
log.reason(f"Starting pipeline for dashboard '{title}'", payload={"dash_id": dash_id})
try:
exported_content, _ = from_c.export_dashboard(dash_id)
@@ -291,10 +294,10 @@ class MigrationPlugin(PluginBase):
if not success and replace_db_config:
if task_id:
app_logger.explore("Missing mapping blocks AST transform. Pausing task for user intervention.", extra={"task_id": task_id})
log.explore("Missing mapping blocks AST transform. Pausing task for user intervention.", error="AST transform blocked by missing mappings", payload={"task_id": task_id})
await tm.wait_for_resolution(task_id)
app_logger.reason("Task resumed, re-evaluating mapping states")
log.reason("Task resumed, re-evaluating mapping states")
db = SessionLocal()
try:
src_env_rt = db.query(Environment).filter(Environment.name == from_env_name).first()
@@ -317,12 +320,12 @@ class MigrationPlugin(PluginBase):
)
if success:
app_logger.reason("Pushing transformed ZIP to target Superset")
log.reason("Pushing transformed ZIP to target Superset")
to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug)
migration_result["migrated_dashboards"].append({"id": dash_id, "title": title})
app_logger.reflect("Import successful", extra={"title": title})
log.reflect("Import successful", payload={"title": title})
else:
app_logger.explore("Transformation strictly failed, bypassing ingestion")
log.explore("Transformation strictly failed, bypassing ingestion", error="Dashboard ZIP transformation failed")
migration_log.error(f"Failed to transform ZIP for dashboard {title}")
migration_result["failed_dashboards"].append({
"id": dash_id, "title": title, "error": "Failed to transform ZIP"
@@ -340,7 +343,7 @@ class MigrationPlugin(PluginBase):
if match_alt:
db_name = match_alt.group(1)
app_logger.explore(f"Missing DB password detected during ingestion. Escalating to UI.", extra={"db_name": db_name})
log.explore("Missing DB password detected during ingestion. Escalating to UI.", error="Missing DB password for import", payload={"db_name": db_name})
if task_id:
tm.await_input(task_id, {
@@ -354,15 +357,15 @@ class MigrationPlugin(PluginBase):
passwords = task.params.get("passwords", {})
if passwords:
app_logger.reason(f"Retrying import for {title} with injected credentials")
log.reason(f"Retrying import for {title} with injected credentials")
to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug, passwords=passwords)
migration_result["migrated_dashboards"].append({"id": dash_id, "title": title})
app_logger.reflect("Password injection unblocked import")
log.reflect("Password injection unblocked import")
if "passwords" in task.params:
del task.params["passwords"]
continue
app_logger.explore(f"Catastrophic dashboard ingestion failure: {exc}")
log.explore("Catastrophic dashboard ingestion failure", error=f"Dashboard ingestion failed: {exc}")
migration_result["failed_dashboards"].append({"id": dash_id, "title": title, "error": str(exc)})
if migration_result["failed_dashboards"]:
@@ -370,20 +373,20 @@ class MigrationPlugin(PluginBase):
# Post-Migration ID Mapping Synchronization
try:
app_logger.reason("Executing incremental ID catalog sync on target")
log.reason("Executing incremental ID catalog sync on target")
db_session = SessionLocal()
mapping_service = IdMappingService(db_session)
mapping_service.sync_environment(tgt_env.id, to_c, incremental=True)
db_session.close()
app_logger.reflect("Incremental catalog sync closed out cleanly")
log.reflect("Incremental catalog sync closed out cleanly")
except Exception as sync_exc:
app_logger.explore(f"ID Mapping sync failed, mapping state might be degraded: {sync_exc}")
log.explore("ID Mapping sync failed, mapping state might be degraded", error=f"ID Mapping sync error: {sync_exc}")
app_logger.reflect("Migration cycle fully resolved", extra={"result": migration_result})
log.reflect("Migration cycle fully resolved", payload={"result": migration_result})
return migration_result
except Exception as e:
app_logger.explore(f"Fatal plugin failure: {e}", exc_info=True)
log.explore("Fatal plugin failure", error=f"Plugin execution failed: {e}", exc_info=True)
raise e
# [/DEF:execute:Function]
# [/DEF:MigrationPlugin:Class]

28
backend/src/plugins/storage/plugin.py
View File

@@ -18,8 +18,11 @@ from typing import Dict, Any, List, Optional
from fastapi import UploadFile
from ...core.plugin_base import PluginBase
from ...core.logger import belief_scope, logger
from ...core.logger import belief_scope
from ...core.cot_logger import MarkerLogger
from ...models.storage import StoredFile, FileCategory
log = MarkerLogger("StoragePlugin")
from ...dependencies import get_config_manager
from ...core.task_manager.context import TaskContext
# [/SECTION]
@@ -121,14 +124,7 @@ class StoragePlugin(PluginBase):
# @POST: Task is executed and logged.
async def execute(self, params: Dict[str, Any], context: Optional[TaskContext] = None):
with belief_scope("StoragePlugin:execute"):
# Use TaskContext logger if available, otherwise fall back to app logger
log = context.logger if context else logger
# Create sub-loggers for different components
storage_log = log.with_source("storage") if context else log
log.with_source("filesystem") if context else log
storage_log.info(f"Executing with params: {params}")
log.reason(f"Executing with params: {params}")
# [/DEF:execute:Function]
# [DEF:get_storage_root:Function]
@@ -172,7 +168,7 @@ class StoragePlugin(PluginBase):
# Clean up any double slashes or leading/trailing slashes for relative path
return os.path.normpath(resolved).strip("/")
except KeyError as e:
logger.warning(f"[StoragePlugin][Coherence:Failed] Missing variable for path resolution: {e}")
log.explore("Missing variable for path resolution", error=str(e))
# Fallback to literal pattern if formatting fails partially (or handle as needed)
return pattern.replace("{", "").replace("}", "")
# [/DEF:resolve_path:Function]
@@ -189,7 +185,7 @@ class StoragePlugin(PluginBase):
# Use singular name for consistency with BackupPlugin and GitService
path = root / category.value
path.mkdir(parents=True, exist_ok=True)
logger.debug(f"[StoragePlugin][Action] Ensured directory: {path}")
log.reason(f"Ensured directory: {path}")
# [/DEF:ensure_directories:Function]
# [DEF:validate_path:Function]
@@ -203,7 +199,7 @@ class StoragePlugin(PluginBase):
try:
resolved.relative_to(root)
except ValueError:
logger.error(f"[StoragePlugin][Coherence:Failed] Path traversal detected: {resolved} is not under {root}")
log.explore(f"Path traversal detected: {resolved} is not under {root}", error="Path outside storage root")
raise ValueError("Access denied: Path is outside of storage root.")
return resolved
# [/DEF:validate_path:Function]
@@ -224,8 +220,8 @@ class StoragePlugin(PluginBase):
) -> List[StoredFile]:
with belief_scope("StoragePlugin:list_files"):
root = self.get_storage_root()
logger.info(
f"[StoragePlugin][Action] Listing files in root: {root}, category: {category}, subpath: {subpath}, recursive: {recursive}"
log.reason(
f"Listing files in root: {root}, category: {category}, subpath: {subpath}, recursive: {recursive}"
)
files = []
@@ -261,7 +257,7 @@ class StoragePlugin(PluginBase):
if not target_dir.exists():
continue
logger.debug(f"[StoragePlugin][Action] Scanning directory: {target_dir}")
log.reason(f"Scanning directory: {target_dir}")
if recursive:
for current_root, dirs, filenames in os.walk(target_dir):
@@ -361,7 +357,7 @@ class StoragePlugin(PluginBase):
shutil.rmtree(full_path)
else:
full_path.unlink()
logger.info(f"[StoragePlugin][Action] Deleted: {full_path}")
log.reflect(f"Deleted: {full_path}")
else:
raise FileNotFoundError(f"Item {path} not found")
# [/DEF:delete_file:Function]

45
backend/src/plugins/translate/dictionary.py
View File

@@ -17,7 +17,8 @@ import re
from typing import Any, Dict, List, Optional, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import func
from ...core.logger import logger, belief_scope
from ...core.cot_logger import MarkerLogger
from ...core.logger import belief_scope
from ...models.translate import (
TerminologyDictionary,
DictionaryEntry,
@@ -26,6 +27,8 @@ from ...models.translate import (
)
from ._utils import _normalize_term, _detect_delimiter
log = MarkerLogger("DictionaryManager")
# #region DictionaryManager [C:5] [TYPE Class] [SEMANTICS dictionary,manager,crud,import]
# @BRIEF Manages terminology dictionaries and their entries with referential integrity, import, and batch filtering.
@@ -45,7 +48,7 @@ class DictionaryManager:
is_active: bool = True,
) -> TerminologyDictionary:
with belief_scope("DictionaryManager.create_dictionary"):
logger.reason("Creating dictionary", {"name": name, "source": source_dialect, "target": target_dialect})
log.reason("Creating dictionary", payload={"name": name, "source": source_dialect, "target": target_dialect})
dictionary = TerminologyDictionary(
name=name,
description=description,
@@ -57,7 +60,7 @@ class DictionaryManager:
db.add(dictionary)
db.commit()
db.refresh(dictionary)
logger.reflect("Dictionary created", {"id": dictionary.id})
log.reflect("Dictionary created", payload={"id": dictionary.id})
return dictionary
# #endregion DictionaryManager.create_dictionary
@@ -74,7 +77,7 @@ class DictionaryManager:
dictionary = db.query(TerminologyDictionary).filter(TerminologyDictionary.id == dict_id).first()
if not dictionary:
raise ValueError(f"Dictionary not found: {dict_id}")
logger.reason("Updating dictionary", {"id": dict_id})
log.reason("Updating dictionary", payload={"id": dict_id})
if name is not None:
dictionary.name = name
if description is not None:
@@ -87,7 +90,7 @@ class DictionaryManager:
dictionary.is_active = is_active
db.commit()
db.refresh(dictionary)
logger.reflect("Dictionary updated", {"id": dictionary.id})
log.reflect("Dictionary updated", payload={"id": dictionary.id})
return dictionary
# #endregion DictionaryManager.update_dictionary
@@ -119,19 +122,19 @@ class DictionaryManager:
)
if attached_jobs > 0:
logger.explore("Delete blocked: dictionary attached to active jobs", {"dict_id": dict_id, "jobs": attached_jobs})
log.explore("Delete blocked: dictionary attached to active jobs", payload={"dict_id": dict_id, "jobs": attached_jobs}, error=f"Dictionary attached to {attached_jobs} active job(s)")
raise ValueError(
f"Cannot delete dictionary attached to {attached_jobs} active/scheduled job(s). "
"Remove dictionary associations from jobs first."
)
logger.reason("Deleting dictionary", {"id": dict_id})
log.reason("Deleting dictionary", payload={"id": dict_id})
# Delete entries and job-dictionary links first
db.query(DictionaryEntry).filter(DictionaryEntry.dictionary_id == dict_id).delete()
db.query(TranslationJobDictionary).filter(TranslationJobDictionary.dictionary_id == dict_id).delete()
db.delete(dictionary)
db.commit()
logger.reflect("Dictionary deleted", {"id": dict_id})
log.reflect("Dictionary deleted", payload={"id": dict_id})
# #endregion DictionaryManager.delete_dictionary
# #region DictionaryManager.get_dictionary [C:2] [TYPE Function] [SEMANTICS dictionary,get]
@@ -185,7 +188,7 @@ class DictionaryManager:
f"(id={existing.id}). Use overwrite or keep_existing conflict mode."
)
logger.reason("Adding dictionary entry", {"dict_id": dict_id, "term": source_term})
log.reason("Adding dictionary entry", payload={"dict_id": dict_id, "term": source_term})
entry = DictionaryEntry(
dictionary_id=dict_id,
source_term=source_term.strip(),
@@ -196,7 +199,7 @@ class DictionaryManager:
db.add(entry)
db.commit()
db.refresh(entry)
logger.reflect("Entry added", {"entry_id": entry.id})
log.reflect("Entry added", payload={"entry_id": entry.id})
return entry
# #endregion DictionaryManager.add_entry
@@ -213,7 +216,7 @@ class DictionaryManager:
if not entry:
raise ValueError(f"Entry not found: {entry_id}")
logger.reason("Editing dictionary entry", {"entry_id": entry_id})
log.reason("Editing dictionary entry", payload={"entry_id": entry_id})
if source_term is not None:
normalized = _normalize_term(source_term)
# Check uniqueness within the same dictionary
@@ -239,7 +242,7 @@ class DictionaryManager:
entry.context_notes = context_notes.strip() if context_notes else None
db.commit()
db.refresh(entry)
logger.reflect("Entry updated", {"entry_id": entry.id})
log.reflect("Entry updated", payload={"entry_id": entry.id})
return entry
# #endregion DictionaryManager.edit_entry
@@ -251,10 +254,10 @@ class DictionaryManager:
entry = db.query(DictionaryEntry).filter(DictionaryEntry.id == entry_id).first()
if not entry:
raise ValueError(f"Entry not found: {entry_id}")
logger.reason("Deleting dictionary entry", {"entry_id": entry_id})
log.reason("Deleting dictionary entry", payload={"entry_id": entry_id})
db.delete(entry)
db.commit()
logger.reflect("Entry deleted", {"entry_id": entry_id})
log.reflect("Entry deleted", payload={"entry_id": entry_id})
# #endregion DictionaryManager.delete_entry
# #region DictionaryManager.clear_entries [C:2] [TYPE Function] [SEMANTICS dictionary,entry,clear]
@@ -265,10 +268,10 @@ class DictionaryManager:
dictionary = db.query(TerminologyDictionary).filter(TerminologyDictionary.id == dict_id).first()
if not dictionary:
raise ValueError(f"Dictionary not found: {dict_id}")
logger.reason("Clearing all entries", {"dict_id": dict_id})
log.reason("Clearing all entries", payload={"dict_id": dict_id})
deleted = db.query(DictionaryEntry).filter(DictionaryEntry.dictionary_id == dict_id).delete()
db.commit()
logger.reflect("Entries cleared", {"dict_id": dict_id, "count": deleted})
log.reflect("Entries cleared", payload={"dict_id": dict_id, "count": deleted})
return deleted
# #endregion DictionaryManager.clear_entries
@@ -316,7 +319,7 @@ class DictionaryManager:
# Detect delimiter if not specified
if not delimiter:
delimiter = _detect_delimiter(content)
logger.reason("Detected delimiter", {"delimiter": repr(delimiter)})
log.reason("Detected delimiter", payload={"delimiter": repr(delimiter)})
if delimiter not in (",", "\t"):
raise ValueError(f"Unsupported delimiter: {repr(delimiter)}. Use ',' or '\\t'.")
@@ -423,7 +426,7 @@ class DictionaryManager:
if not preview_only:
db.commit()
logger.reflect("Import complete", {
log.reflect("Import complete", payload={
"dict_id": dict_id,
"total": result["total"],
"created": result["created"],
@@ -451,7 +454,7 @@ class DictionaryManager:
.all()
)
if not job_dict_links:
logger.reason("No dictionaries attached to job", {"job_id": job_id})
log.reason("No dictionaries attached to job", payload={"job_id": job_id})
return []
dict_ids = [jd.dictionary_id for jd in job_dict_links]
@@ -522,7 +525,7 @@ class DictionaryManager:
dict_priority = {d_id: idx for idx, d_id in enumerate(dict_ids)}
matched.sort(key=lambda m: (dict_priority.get(m["dictionary_id"], 999), m["text_index"]))
logger.reflect("Batch filter match complete", {
log.reflect("Batch filter match complete", payload={
"job_id": job_id,
"source_texts": len(source_texts),
"matches": len(matched),
@@ -625,7 +628,7 @@ class DictionaryManager:
result["message"] = f"Entry created for '{source_term}' -> '{corrected_target_term}'"
db.commit()
logger.reflect("Correction processed", result)
log.reflect("Correction processed", payload=result)
return result
# #endregion DictionaryManager.submit_correction

15
backend/src/plugins/translate/events.py
View File

@@ -16,9 +16,12 @@ from sqlalchemy.orm import Session
from datetime import datetime, timezone, timedelta
import uuid
from ...core.logger import logger, belief_scope
from ...core.cot_logger import MarkerLogger
from ...core.logger import belief_scope
from ...models.translate import TranslationEvent, MetricSnapshot
log = MarkerLogger("EventLog")
# Terminal events per run — exactly one of these must exist for a completed run.
TERMINAL_EVENT_TYPES = {"RUN_COMPLETED", "RUN_FAILED", "RUN_CANCELLED"}
# All valid event types for validation.
@@ -133,7 +136,7 @@ class TranslationEventLog:
run_id=run_id,
created_by=created_by,
)
logger.reason(f"Event logged: {event_type}", {
log.reason("Event logged", payload={
"event_id": event.id,
"job_id": job_id,
"run_id": run_id,
@@ -195,7 +198,7 @@ class TranslationEventLog:
) -> Dict[str, Any]:
with belief_scope("TranslationEventLog.prune_expired"):
cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
logger.reason("Pruning expired events", {
log.reason("Pruning expired events", payload={
"cutoff": cutoff.isoformat(),
"retention_days": retention_days,
})
@@ -207,7 +210,7 @@ class TranslationEventLog:
total_expired = expired_query.count()
if total_expired == 0:
logger.reflect("No expired events to prune", {})
log.reflect("No expired events to prune", payload={})
return {"pruned": 0, "snapshot_id": None}
# Create MetricSnapshot before pruning
@@ -240,11 +243,11 @@ class TranslationEventLog:
).delete(synchronize_session=False)
self.db.flush()
pruned += len(ids)
logger.reason("Pruned batch", {"batch_size": len(ids), "total_pruned": pruned})
log.reason("Pruned batch", payload={"batch_size": len(ids), "total_pruned": pruned})
self.db.commit()
logger.reflect("Pruning complete", {
log.reflect("Pruning complete", payload={
"pruned": pruned,
"snapshot_id": snapshot_id,
})

47
backend/src/plugins/translate/executor.py
View File

@@ -24,6 +24,7 @@ from typing import Any, Dict, List, Optional, Set, Tuple, Callable
from sqlalchemy.orm import Session
from ...core.logger import logger, belief_scope
from ...core.cot_logger import MarkerLogger
from ...core.config_manager import ConfigManager
from ...models.translate import (
TranslationJob,
@@ -39,6 +40,7 @@ from ...core.superset_client import SupersetClient
from .dictionary import DictionaryManager
from .preview import DEFAULT_EXECUTION_PROMPT_TEMPLATE
log = MarkerLogger("TranslationExecutor")
# #region MAX_RETRIES_PER_BATCH [C:1] [TYPE Constant] [SEMANTICS translate,executor,retry]
MAX_RETRIES_PER_BATCH = 3
@@ -83,7 +85,7 @@ class TranslationExecutor:
if not job:
raise ValueError(f"Job '{run.job_id}' not found for run '{run.id}'")
logger.reason("Starting translation execution", {
log.reason("Starting translation execution", payload={
"run_id": run.id,
"job_id": job.id,
"batch_size": job.batch_size,
@@ -103,7 +105,7 @@ class TranslationExecutor:
# Preview-based: fetch rows from the accepted preview session
source_rows = self._fetch_source_rows(job.id, run.id)
if not source_rows:
logger.explore("No source rows to translate", {"run_id": run.id})
log.explore("No source rows to translate", payload={"run_id": run.id}, error="Preview produced 0 source rows for translation")
run.status = "COMPLETED"
run.completed_at = datetime.now(timezone.utc)
self.db.flush()
@@ -119,7 +121,7 @@ class TranslationExecutor:
for i in range(0, total_rows, batch_size)
]
logger.reason(f"Processing {len(batches)} batches", {
log.reason(f"Processing {len(batches)} batches", payload={
"run_id": run.id,
"total_rows": total_rows,
"batch_size": batch_size,
@@ -163,7 +165,7 @@ class TranslationExecutor:
run.completed_at = datetime.now(timezone.utc)
self.db.flush()
logger.reflect("Translation execution complete", {
log.reflect("Translation execution complete", payload={
"run_id": run.id,
"status": run.status,
"total": total_rows,
@@ -193,7 +195,7 @@ class TranslationExecutor:
.first()
)
if not session:
logger.explore("No accepted preview session found", {"job_id": job_id})
log.explore("No accepted preview session found", error="Preview session has no accepted rows", payload={"job_id": job_id})
return []
# Fetch APPROVED or all records from the session
@@ -216,7 +218,7 @@ class TranslationExecutor:
"source_data": rec.source_data or {},
})
logger.reason(f"Fetched {len(source_rows)} source rows from preview", {
log.reason(f"Fetched {len(source_rows)} source rows from preview", payload={
"run_id": run_id,
"session_id": session.id,
})
@@ -278,7 +280,8 @@ class TranslationExecutor:
headers={"Content-Type": "application/json"},
)
except Exception as e:
logger.explore("Chart data API failed during full fetch", {
log.explore("Chart data API failed during full fetch", error="Chart data API error",
payload={
"offset": len(all_rows),
"error": str(e),
})
@@ -292,7 +295,7 @@ class TranslationExecutor:
break # No more data
all_rows.extend(rows)
logger.reason(f"Fetched {len(rows)} rows (total: {len(all_rows)})", {
log.reason(f"Fetched {len(rows)} rows (total: {len(all_rows)})", payload={
"offset": len(all_rows) - len(rows),
})
@@ -319,7 +322,7 @@ class TranslationExecutor:
"approved_translation": None,
})
logger.reason(f"Prepared {len(source_rows)} source rows for full translation", {
log.reason(f"Prepared {len(source_rows)} source rows for full translation", payload={
"job_id": job.id,
})
return source_rows
@@ -416,7 +419,7 @@ class TranslationExecutor:
self.db.flush()
batch_latency = int((time.monotonic() - batch_start) * 1000)
logger.reason(f"Batch {batch_index} complete", {
log.reason(f"Batch {batch_index} complete", payload={
"batch_id": batch_id,
"latency_ms": batch_latency,
**result,
@@ -490,7 +493,8 @@ class TranslationExecutor:
except Exception as e:
last_error = str(e)
retries += 1
logger.explore(f"LLM call failed (attempt {attempt})", {
log.explore("LLM call failed", error="LLM call failed, retrying",
payload={
"batch_id": batch_id,
"error": last_error,
"attempt": attempt,
@@ -498,7 +502,8 @@ class TranslationExecutor:
if attempt < MAX_RETRIES_PER_BATCH:
time.sleep(2 ** attempt) # Exponential backoff
else:
logger.explore("LLM call exhausted retries", {
log.explore("LLM call exhausted retries", error="LLM retries exhausted",
payload={
"batch_id": batch_id,
"last_error": last_error,
})
@@ -527,7 +532,8 @@ class TranslationExecutor:
translations = self._parse_llm_response(llm_response, len(batch_rows))
except ValueError as e:
# Parse failure — mark all rows as SKIPPED
logger.explore("LLM response parse failed", {
log.explore("LLM response parse failed", error="Failed to parse LLM JSON response",
payload={
"batch_id": batch_id,
"error": str(e),
"response_preview": llm_response[:500] if llm_response else "",
@@ -693,7 +699,7 @@ class TranslationExecutor:
if provider_type in ("openai", "openai_compatible"):
payload["response_format"] = {"type": "json_object"}
logger.reason(
log.reason(
f"LLM request model={payload.get('model')} "
f"provider_type={provider_type} "
f"response_format={'yes' if 'response_format' in payload else 'no'} "
@@ -701,11 +707,11 @@ class TranslationExecutor:
)
response = http_requests.post(url, headers=headers, json=payload, timeout=180)
if not response.ok:
logger.explore(
f"LLM API error status={response.status_code} "
f"model={payload.get('model')} "
f"body={response.text[:2000]}"
)
log.explore("LLM API error", error=f"LLM API returned status {response.status_code}", payload={
"status_code": response.status_code,
"model": payload.get('model'),
"body": response.text[:2000],
})
response.raise_for_status()
data = response.json()
@@ -717,7 +723,8 @@ class TranslationExecutor:
if not content:
# Log full response for diagnostics
finish_reason = choices[0].get("finish_reason", "unknown")
logger.explore("LLM returned empty content", {
log.explore("LLM returned empty content", error="Empty response from LLM",
payload={
"finish_reason": finish_reason,
"model": payload.get("model"),
"prompt_len": len(prompt),

38
backend/src/plugins/translate/orchestrator.py
View File

@@ -26,6 +26,7 @@ from typing import Any, Dict, List, Optional, Callable, Tuple
from sqlalchemy.orm import Session
from ...core.logger import logger, belief_scope
from ...core.cot_logger import MarkerLogger
from ...core.config_manager import ConfigManager
from ...models.translate import (
TranslationJob,
@@ -41,6 +42,7 @@ from .superset_executor import SupersetSqlLabExecutor
from .events import TranslationEventLog
from ..translate.service import TranslateJobService
log = MarkerLogger("TranslationOrchestrator")
# #region TranslationOrchestrator [C:5] [TYPE Class] [SEMANTICS translate,orchestrator]
# @BRIEF Coordinates full translation run lifecycle: validation, execution, SQL generation, Superset submission, event logging.
@@ -81,7 +83,7 @@ class TranslationOrchestrator:
raise ValueError(f"Translation job '{job_id}' not found")
self._job = job
logger.reason("Starting translation run", {
log.reason("Starting translation run", payload={
"job_id": job_id,
"is_scheduled": is_scheduled,
})
@@ -156,7 +158,7 @@ class TranslationOrchestrator:
self.db.commit()
self.db.refresh(run)
logger.reflect("Run created", {
log.reflect("Run created", payload={
"run_id": run.id,
"job_id": job_id,
"status": run.status,
@@ -189,7 +191,7 @@ class TranslationOrchestrator:
f"Run must be in PENDING status."
)
logger.reason("Executing run", {
log.reason("Executing run", payload={
"run_id": run.id,
"job_id": job.id,
"skip_insert": skip_insert,
@@ -212,9 +214,8 @@ class TranslationOrchestrator:
try:
run = executor.execute_run(run, llm_progress_callback=None, full_translation=full_translation)
except Exception as e:
logger.explore("Translation execution failed", {
log.explore("Translation execution failed", error=str(e), payload={
"run_id": run.id,
"error": str(e),
})
run.status = "FAILED"
run.error_message = f"Translation execution failed: {e}"
@@ -291,7 +292,7 @@ class TranslationOrchestrator:
self.db.commit()
self.db.refresh(run)
logger.reflect("Run execution complete", {
log.reflect("Run execution complete", payload={
"run_id": run.id,
"status": run.status,
"insert_status": run.insert_status,
@@ -322,10 +323,10 @@ class TranslationOrchestrator:
)
if not records:
logger.reason("No successful records to insert", {"run_id": run.id})
log.reason("No successful records to insert", payload={"run_id": run.id})
return {"status": "skipped", "reason": "no_records", "query_id": None}
logger.reason(f"Generating SQL for {len(records)} records", {
log.reason(f"Generating SQL for {len(records)} records", payload={
"run_id": run.id,
"dialect": job.database_dialect or job.target_dialect,
})
@@ -371,11 +372,11 @@ class TranslationOrchestrator:
upsert_strategy=job.upsert_strategy or "MERGE",
)
except ValueError as e:
logger.explore("SQL generation failed", {"error": str(e)})
log.explore("SQL generation failed", error=str(e))
return {"status": "failed", "error_message": str(e), "query_id": None}
# Log insert phase start
logger.reason("Insert SQL generated", {
log.reason("Insert SQL generated", payload={
"run_id": run.id,
"sql_preview": sql[:500],
"sql_length": len(sql),
@@ -404,7 +405,7 @@ class TranslationOrchestrator:
except (ValueError, TypeError):
# Could be a UUID — pass as-is, executor will resolve it
target_db_id = job.target_database_id
logger.reason("target_database_id is not an integer, will resolve as UUID", {
log.reason("target_database_id is not an integer, will resolve as UUID", payload={
"target_database_id": job.target_database_id,
})
executor = SupersetSqlLabExecutor(self.config_manager, env_id, database_id=target_db_id)
@@ -414,9 +415,8 @@ class TranslationOrchestrator:
poll_interval_seconds=2.0,
)
except Exception as e:
logger.explore("Superset SQL submission failed", {
log.explore("Superset SQL submission failed", error=str(e), payload={
"run_id": run.id,
"error": str(e),
})
result = {"status": "failed", "error_message": str(e), "query_id": None}
@@ -488,7 +488,7 @@ class TranslationOrchestrator:
"Run and accept a preview before executing a manual translation run."
)
logger.reason("Preconditions validated", {
log.reason("Preconditions validated", payload={
"job_id": job.id,
"is_scheduled": is_scheduled,
})
@@ -526,7 +526,7 @@ class TranslationOrchestrator:
if not failed_batches:
raise ValueError(f"No failed batches found for run '{run_id}'")
logger.reason("Retrying failed batches", {
log.reason("Retrying failed batches", payload={
"run_id": run_id,
"batch_count": len(failed_batches),
})
@@ -602,7 +602,7 @@ class TranslationOrchestrator:
)
self.db.commit()
logger.reflect("Retry complete", {
log.reflect("Retry complete", payload={
"run_id": run_id,
"status": run.status,
})
@@ -628,7 +628,7 @@ class TranslationOrchestrator:
raise ValueError(f"Job '{run.job_id}' not found")
self._job = job
logger.reason("Retrying insert phase", {
log.reason("Retrying insert phase", payload={
"run_id": run_id,
})
@@ -652,7 +652,7 @@ class TranslationOrchestrator:
self.db.commit()
self.db.refresh(run)
logger.reflect("Insert retry complete", {
log.reflect("Insert retry complete", payload={
"run_id": run_id,
"insert_status": run.insert_status,
})
@@ -692,7 +692,7 @@ class TranslationOrchestrator:
self.db.commit()
self.db.refresh(run)
logger.reflect("Run cancelled", {
log.reflect("Run cancelled", payload={
"run_id": run_id,
})
return run

60
backend/src/plugins/translate/preview.py
View File

@@ -25,7 +25,8 @@ import json
import hashlib
import re
from ...core.logger import logger, belief_scope
from ...core.cot_logger import MarkerLogger
from ...core.logger import belief_scope
from ...core.config_manager import ConfigManager
from ...core.superset_client import SupersetClient
from ...models.translate import (
@@ -38,6 +39,8 @@ from ...services.llm_provider import LLMProviderService
from ...services.llm_prompt_templates import render_prompt
from .dictionary import DictionaryManager
log = MarkerLogger("TranslationPreview")
# #region DEFAULT_EXECUTION_PROMPT_TEMPLATE [C:1] [TYPE Constant] [SEMANTICS translate,prompt]
DEFAULT_EXECUTION_PROMPT_TEMPLATE: str = (
"Translate the following database content from {source_language} to {target_language}.\n\n"
@@ -137,7 +140,7 @@ class TranslationPreview:
env_id: Optional[str] = None,
) -> Dict[str, Any]:
with belief_scope("TranslationPreview.preview_rows"):
logger.reason("Starting preview for job", {"job_id": job_id, "sample_size": sample_size})
log.reason("Starting preview for job", payload={"job_id": job_id, "sample_size": sample_size})
# 1. Load job
job = self.db.query(TranslationJob).filter(TranslationJob.id == job_id).first()
@@ -153,7 +156,7 @@ class TranslationPreview:
dict_snapshot_hash = self._compute_dict_snapshot_hash(job_id)
# 3. Fetch sample rows from Superset
logger.reason("Fetching sample rows from Superset", {
log.reason("Fetching sample rows from Superset", payload={
"datasource_id": job.source_datasource_id,
"sample_size": sample_size,
"translation_column": job.translation_column,
@@ -167,12 +170,12 @@ class TranslationPreview:
raise ValueError("No rows returned from datasource for preview")
actual_row_count = len(source_rows)
logger.reason(f"Fetched {actual_row_count} sample row(s)")
log.reason(f"Fetched {actual_row_count} sample row(s)")
# Debug: log first row keys and translation column value
if source_rows:
first_row = source_rows[0]
logger.reason(
log.reason(
f"First source row keys={list(first_row.keys())} "
f"translation_col={job.translation_column} "
f"val='{first_row.get(job.translation_column, '')}'"
@@ -246,7 +249,7 @@ class TranslationPreview:
total_est_cost = TokenEstimator.estimate_cost(total_est_tokens)
# 8. Call LLM
logger.reason("Calling LLM for preview translation", {
log.reason("Calling LLM for preview translation", payload={
"provider_id": job.provider_id,
"row_count": actual_row_count,
"estimated_tokens": sample_total_tokens,
@@ -330,7 +333,7 @@ class TranslationPreview:
"dict_snapshot_hash": dict_snapshot_hash,
}
logger.reflect("Preview completed", {
log.reflect("Preview completed", payload={
"session_id": session.id,
"row_count": actual_row_count,
"sample_cost": sample_cost,
@@ -393,7 +396,7 @@ class TranslationPreview:
self.db.commit()
self.db.refresh(record)
logger.reason(f"Preview row {action}d", {
log.reason(f"Preview row {action}d", payload={
"row_id": row_id,
"session_id": session.id,
"status": record.status,
@@ -431,7 +434,7 @@ class TranslationPreview:
self.db.commit()
self.db.refresh(session)
logger.reason("Preview session accepted", {
log.reason("Preview session accepted", payload={
"session_id": session.id,
"job_id": job_id,
})
@@ -521,13 +524,15 @@ class TranslationPreview:
None,
)
if not env_config:
logger.explore("Could not find environment for datasource", {
log.explore("Could not find environment for datasource", error="Environment not found for datasource",
payload={
"env_id": target_env_id,
})
# Fallback: try first environment
if environments:
env_config = environments[0]
logger.explore("Falling back to first available environment", {
log.explore("Falling back to first available environment", error="Environment fallback used",
payload={
"env_id": env_config.id,
})
else:
@@ -572,17 +577,17 @@ class TranslationPreview:
headers={"Content-Type": "application/json"},
)
except Exception as e:
logger.explore("Chart data API failed", {"error": str(e)})
log.explore("Chart data API failed", error=str(e))
raise ValueError(f"Failed to fetch sample data from Superset: {e}")
# Parse response
rows = self._extract_data_rows(response)
logger.reason(f"Extracted {len(rows)} data row(s)")
log.reason(f"Extracted {len(rows)} data row(s)")
# Debug: log first row keys and translation column value
if rows:
first_row = rows[0]
logger.reason(
log.reason(
f"Row keys={list(first_row.keys())} "
f"target_col={job.translation_column} "
f"val='{first_row.get(job.translation_column, '')}'"
@@ -658,7 +663,7 @@ class TranslationPreview:
else:
raise ValueError(f"Unsupported provider type '{provider_type}' for preview")
logger.reason("LLM call completed", {
log.reason("LLM call completed", payload={
"provider_id": job.provider_id,
"model": model,
"response_length": len(response_text),
@@ -701,7 +706,7 @@ class TranslationPreview:
if provider_type in ("openai", "openai_compatible"):
payload["response_format"] = {"type": "json_object"}
logger.reason(
log.reason(
f"LLM request model={payload.get('model')} "
f"provider_type={provider_type} "
f"response_format={'yes' if 'response_format' in payload else 'no'} "
@@ -709,11 +714,11 @@ class TranslationPreview:
)
response = http_requests.post(url, headers=headers, json=payload, timeout=120)
if not response.ok:
logger.explore(
f"LLM API error status={response.status_code} "
f"model={payload.get('model')} "
f"body={response.text[:2000]}"
)
log.explore("LLM API error", error=f"LLM API returned status {response.status_code}", payload={
"status_code": response.status_code,
"model": payload.get('model'),
"body": response.text[:2000],
})
response.raise_for_status()
data = response.json()
@@ -736,7 +741,7 @@ class TranslationPreview:
@staticmethod
def _parse_llm_response(response_text: str, expected_count: int) -> Dict[str, str]:
with belief_scope("TranslationPreview._parse_llm_response"):
logger.reason(f"Raw LLM response length={len(response_text)} preview={response_text[:500]}")
log.reason(f"Raw LLM response length={len(response_text)} preview={response_text[:500]}")
try:
data = json.loads(response_text)
@@ -754,7 +759,7 @@ class TranslationPreview:
rows = data.get("rows", [])
if not isinstance(rows, list):
logger.explore(f"LLM response has no 'rows' array, keys={list(data.keys())} text_preview={response_text[:300]}")
log.explore("LLM response has no 'rows' array", error="LLM response missing 'rows' key")
raise ValueError("LLM response missing 'rows' array")
translations: Dict[str, str] = {}
@@ -765,10 +770,11 @@ class TranslationPreview:
translations[row_id] = translation
if len(translations) < expected_count:
logger.explore(
f"LLM returned fewer translations expected={expected_count} "
f"got={len(translations)} response_preview={response_text[:600]}"
)
log.explore("LLM returned fewer translations than expected", error=f"Expected {expected_count} translations, got {len(translations)}", payload={
"expected_count": expected_count,
"actual_count": len(translations),
"response_preview": response_text[:600],
})
return translations
# #endregion _parse_llm_response

33
backend/src/plugins/translate/scheduler.py
View File

@@ -20,11 +20,14 @@ from typing import Any, Dict, List, Optional
from sqlalchemy.orm import Session
from ...core.logger import logger, belief_scope
from ...core.cot_logger import MarkerLogger
from ...core.logger import belief_scope
from ...core.config_manager import ConfigManager
from ...models.translate import TranslationSchedule, TranslationJob, TranslationRun
from .events import TranslationEventLog
log = MarkerLogger("TranslationScheduler")
# #region TranslationScheduler [C:5] [TYPE Class] [SEMANTICS translate,scheduler,crud]
# @BRIEF CRUD for TranslationSchedule rows + APScheduler registration wrappers.
@@ -82,7 +85,7 @@ class TranslationScheduler:
created_by=self.current_user,
)
logger.reflect("Schedule created", {"schedule_id": schedule.id, "job_id": job_id})
log.reflect("Schedule created", payload={"schedule_id": schedule.id, "job_id": job_id})
return schedule
# #endregion create_schedule
@@ -127,7 +130,7 @@ class TranslationScheduler:
created_by=self.current_user,
)
logger.reflect("Schedule updated", {"schedule_id": schedule.id, "job_id": job_id})
log.reflect("Schedule updated", payload={"schedule_id": schedule.id, "job_id": job_id})
return schedule
# #endregion update_schedule
@@ -155,7 +158,7 @@ class TranslationScheduler:
created_by=self.current_user,
)
logger.reflect("Schedule deleted", {"schedule_id": schedule_id, "job_id": job_id})
log.reflect("Schedule deleted", payload={"schedule_id": schedule_id, "job_id": job_id})
# #endregion delete_schedule
# #region set_schedule_active [C:4] [TYPE Function] [SEMANTICS translate,schedule,activate]
@@ -176,7 +179,7 @@ class TranslationScheduler:
self.db.commit()
self.db.refresh(schedule)
logger.reflect("Schedule active state set", {
log.reflect("Schedule active state set", payload={
"schedule_id": schedule.id,
"job_id": job_id,
"is_active": is_active,
@@ -218,7 +221,7 @@ class TranslationScheduler:
tz = ZoneInfo(timezone_str)
trigger = CronTrigger.from_crontab(cron_expression, timezone=tz)
except (ValueError, KeyError) as e:
logger.warning(f"[get_next_executions] Invalid cron: {e}")
log.explore("Invalid cron", error=str(e))
return []
now = datetime.now(tz)
@@ -263,10 +266,10 @@ def execute_scheduled_translation(
TranslationSchedule.job_id == job_id,
).first()
if not schedule or not schedule.is_active:
logger.info(f"[scheduled_translation] Schedule inactive/missing: {schedule_id}")
log.reason("Schedule inactive/missing", payload={"schedule_id": schedule_id})
return
logger.reason("Scheduled translation triggered", {
log.reason("Scheduled translation triggered", payload={
"schedule_id": schedule_id,
"job_id": job_id,
})
@@ -281,7 +284,8 @@ def execute_scheduled_translation(
.first()
)
if active_run:
logger.explore("Skipping scheduled run — concurrent run in progress", {
log.explore("Skipping scheduled run — concurrent run in progress", error="Concurrent run detected",
payload={
"job_id": job_id,
"existing_run_id": active_run.id,
})
@@ -308,7 +312,7 @@ def execute_scheduled_translation(
age = datetime.now(timezone.utc) - most_recent.created_at
if age > timedelta(days=90):
baseline_expired = True
logger.reason("Baseline expired — full translation", {
log.reason("Baseline expired — full translation", payload={
"job_id": job_id,
"last_run": most_recent.created_at.isoformat(),
"age_days": age.days,
@@ -339,10 +343,7 @@ def execute_scheduled_translation(
orch.execute_run(run)
run.insert_status = run.insert_status or "succeeded"
except Exception as exec_err:
logger.explore("Scheduled translation execution failed", {
"run_id": run.id,
"error": str(exec_err),
})
log.explore("Scheduled translation execution failed", error=str(exec_err))
# Leave schedule enabled — schedule continues on failure
run.status = "FAILED"
run.error_message = str(exec_err)
@@ -352,13 +353,13 @@ def execute_scheduled_translation(
schedule.last_run_at = datetime.now(timezone.utc)
db.commit()
logger.reflect("Scheduled translation complete", {
log.reflect("Scheduled translation complete", payload={
"schedule_id": schedule_id,
"run_id": run.id,
"status": run.status,
})
except Exception as e:
logger.error(f"[scheduled_translation] Unexpected error: {e}")
log.explore("Unexpected error", error=str(e))
finally:
db.close()
# #endregion execute_scheduled_translation

49
backend/src/plugins/translate/service.py
View File

@@ -18,6 +18,7 @@ from datetime import datetime, timezone
import uuid
from ...core.logger import logger
from ...core.cot_logger import MarkerLogger
from ...core.config_manager import ConfigManager
from ...core.superset_client import SupersetClient
from ...models.translate import TranslationJob, TranslationJobDictionary
@@ -29,6 +30,8 @@ from ...schemas.translate import (
DatasourceColumnResponse,
)
log = MarkerLogger("TranslateJobService")
# Supported database dialects for translation
SUPPORTED_DIALECTS = {
"postgresql", "mysql", "clickhouse", "sqlite", "mssql",
@@ -44,7 +47,7 @@ SUPPORTED_DIALECTS = {
# @SIDE_EFFECT None — pure data extraction.
def get_dialect_from_database(database_record: Dict[str, Any]) -> str:
"""Extract and validate dialect from Superset database record."""
logger.reason("Extracting dialect from database record", {
log.reason("Extracting dialect from database record", payload={
"has_backend": bool(database_record.get("backend") or database_record.get("engine")),
})
backend = (
@@ -54,7 +57,7 @@ def get_dialect_from_database(database_record: Dict[str, Any]) -> str:
).lower().strip()
if not backend:
logger.explore("Could not determine database dialect", {
log.explore("Could not determine database dialect", error="No backend/engine field in database record", payload={
"record_keys": list(database_record.keys()),
})
raise ValueError("Could not determine database dialect from connection")
@@ -82,12 +85,12 @@ def get_dialect_from_database(database_record: Dict[str, Any]) -> str:
normalized = dialect_map.get(backend, backend)
if normalized not in SUPPORTED_DIALECTS:
logger.explore("Unsupported dialect", {"backend": backend, "normalized": normalized})
log.explore("Unsupported dialect", payload={"backend": backend, "normalized": normalized}, error=f"Unsupported database dialect: {backend}")
raise ValueError(
f"Unsupported database dialect: '{backend}'. "
f"Supported dialects: {', '.join(sorted(SUPPORTED_DIALECTS))}"
)
logger.reflect("Dialect validated", {"dialect": normalized})
log.reflect("Dialect validated", payload={"dialect": normalized})
return normalized
# #endregion get_dialect_from_database
@@ -103,7 +106,7 @@ def fetch_datasource_metadata(
config_manager: ConfigManager,
) -> Tuple[List[Dict[str, Any]], str]:
"""Fetch column metadata and database dialect for a datasource from Superset."""
logger.reason("Fetching datasource metadata", {
log.reason("Fetching datasource metadata", payload={
"dataset_id": dataset_id,
"env_id": env_id,
})
@@ -111,7 +114,7 @@ def fetch_datasource_metadata(
environments = config_manager.get_environments()
env_config = next((e for e in environments if e.id == env_id), None)
if not env_config:
logger.explore("Environment not found", {"env_id": env_id})
log.explore("Environment not found", payload={"env_id": env_id}, error=f"Environment {env_id} not configured")
raise ValueError(f"Superset environment '{env_id}' not found in configuration")
# Create Superset client and fetch dataset detail
@@ -148,10 +151,10 @@ def fetch_datasource_metadata(
else:
raise ValueError("No database information available for this datasource")
except Exception as e:
logger.explore("Could not fetch database dialect", {"error": str(e)})
log.explore("Could not fetch database dialect", error=str(e))
raise ValueError(f"Could not determine database dialect: {e}")
logger.reflect("Metadata fetched", {"columns": len(columns), "dialect": dialect})
log.reflect("Metadata fetched", payload={"columns": len(columns), "dialect": dialect})
return columns, dialect
# #endregion fetch_datasource_metadata
@@ -213,11 +216,11 @@ class TranslateJobService:
# @POST Returns the created TranslationJob with database_dialect cached.
# @SIDE_EFFECT Validates columns via SupersetClient if source_datasource_id is provided; caches database_dialect.
def create_job(self, payload: TranslateJobCreate) -> TranslationJob:
logger.reason("Creating translation job", {"name": payload.name})
log.reason("Creating translation job", payload={"name": payload.name})
# Validate: must have a translation column if datasource is configured
if payload.source_datasource_id and not payload.translation_column:
logger.explore("Missing translation column", {
log.explore("Missing translation column", error="translation_column required when source_datasource_id is set", payload={
"source_datasource_id": payload.source_datasource_id,
})
raise ValueError("A translation column is required when a datasource is selected")
@@ -244,7 +247,7 @@ class TranslateJobService:
)
dialect = detected_dialect
except Exception as e:
logger.explore("Dialect detection failed", {"error": str(e)})
log.explore("Dialect detection failed", error=str(e))
dialect = payload.source_dialect
# Build job instance
@@ -287,7 +290,7 @@ class TranslateJobService:
self.db.commit()
self.db.refresh(job)
logger.reflect("Job created", {"job_id": job.id})
log.reflect("Job created", payload={"job_id": job.id})
return job
# #endregion create_job
@@ -297,7 +300,7 @@ class TranslateJobService:
# @POST Returns the updated TranslationJob with re-detected dialect if datasource changed.
# @SIDE_EFFECT Re-detects database_dialect if source_datasource_id changed.
def update_job(self, job_id: str, payload: TranslateJobUpdate) -> TranslationJob:
logger.reason("Updating translation job", {"job_id": job_id})
log.reason("Updating translation job", payload={"job_id": job_id})
job = self.get_job(job_id)
update_data = payload.model_dump(exclude_unset=True)
@@ -318,7 +321,7 @@ class TranslateJobService:
)
job.database_dialect = detected_dialect
except Exception as e:
logger.explore("Dialect re-detection failed", {"error": str(e)})
log.explore("Dialect re-detection failed", error=str(e))
job.updated_at = datetime.now(timezone.utc)
@@ -337,7 +340,7 @@ class TranslateJobService:
self.db.commit()
self.db.refresh(job)
logger.reflect("Job updated", {"job_id": job_id})
log.reflect("Job updated", payload={"job_id": job_id})
return job
# #endregion update_job
@@ -347,7 +350,7 @@ class TranslateJobService:
# @POST Job and all related TranslationJobDictionary records are deleted.
# @SIDE_EFFECT Deletes TranslationJob and TranslationJobDictionary rows.
def delete_job(self, job_id: str) -> None:
logger.reason("Deleting translation job", {"job_id": job_id})
log.reason("Deleting translation job", payload={"job_id": job_id})
job = self.get_job(job_id)
# Delete dictionary associations
@@ -357,7 +360,7 @@ class TranslateJobService:
self.db.delete(job)
self.db.commit()
logger.reflect("Job deleted", {"job_id": job_id})
log.reflect("Job deleted", payload={"job_id": job_id})
# #endregion delete_job
# #region duplicate_job [C:4] [TYPE Function] [SEMANTICS translate,jobs,duplicate]
@@ -366,7 +369,7 @@ class TranslateJobService:
# @POST Returns the duplicated TranslationJob with status DRAFT.
# @SIDE_EFFECT Creates new TranslationJob and TranslationJobDictionary rows.
def duplicate_job(self, job_id: str, new_name: Optional[str] = None) -> TranslationJob:
logger.reason("Duplicating translation job", {"job_id": job_id, "new_name": new_name})
log.reason("Duplicating translation job", payload={"job_id": job_id, "new_name": new_name})
source = self.get_job(job_id)
# Copy all fields except id, created_at, updated_at, status
@@ -411,7 +414,7 @@ class TranslateJobService:
self.db.commit()
self.db.refresh(new_job)
logger.reflect("Job duplicated", {"new_job_id": new_job.id, "source_job_id": job_id})
log.reflect("Job duplicated", payload={"new_job_id": new_job.id, "source_job_id": job_id})
return new_job
# #endregion duplicate_job
@@ -512,13 +515,13 @@ def get_datasource_columns(
config_manager: ConfigManager,
) -> DatasourceColumnsResponse:
"""Fetch and return column metadata for a given Superset datasource."""
logger.reason("Fetching datasource columns", {"datasource_id": datasource_id, "env_id": env_id})
log.reason("Fetching datasource columns", payload={"datasource_id": datasource_id, "env_id": env_id})
# Find environment config
environments = config_manager.get_environments()
env_config = next((e for e in environments if e.id == env_id), None)
if not env_config:
logger.explore("Environment not found", {"env_id": env_id})
log.explore("Environment not found", payload={"env_id": env_id}, error=f"Environment {env_id} not configured")
raise ValueError(f"Superset environment '{env_id}' not found")
# Create Superset client
@@ -540,7 +543,7 @@ def get_datasource_columns(
db_result = db_record.get("result", db_record) if isinstance(db_record, dict) else db_record
dialect = get_dialect_from_database(db_result)
else:
logger.explore("Could not determine dialect", {"datasource_id": datasource_id})
log.explore("Could not determine dialect", payload={"datasource_id": datasource_id}, error=f"Could not determine dialect for datasource {datasource_id}")
raise ValueError("Could not determine database dialect for this datasource")
# Extract columns
@@ -565,7 +568,7 @@ def get_datasource_columns(
database_dialect=dialect,
columns=columns,
)
logger.reflect("Datasource columns fetched", {
log.reflect("Datasource columns fetched", payload={
"datasource_id": datasource_id,
"columns_count": len(columns),
"dialect": dialect,

12
backend/src/plugins/translate/sql_generator.py
View File

@@ -8,7 +8,11 @@
from typing import Any, Dict, List, Optional, Tuple
from datetime import datetime, timezone
from ...core.logger import logger, belief_scope
from ...core.cot_logger import MarkerLogger
from ...core.logger import belief_scope
log = MarkerLogger("SqlGenerator")
# PostgreSQL/Greenplum dialects that support ON CONFLICT
POSTGRESQL_DIALECTS = {"postgresql", "redshift", "greenplum"}
@@ -240,7 +244,7 @@ class SQLGenerator:
Tuple of (sql_string, row_count).
"""
with belief_scope("SQLGenerator.generate"):
logger.reason("Generating SQL", {
log.reason("Generating SQL", payload={
"dialect": dialect,
"schema": target_schema,
"table": target_table,
@@ -306,7 +310,7 @@ class SQLGenerator:
dialect=dialect,
)
if use_upsert:
logger.reason("ClickHouse UPSERT not supported; using plain INSERT", {
log.reason("ClickHouse UPSERT not supported; using plain INSERT", payload={
"note": "ClickHouse does not support ON CONFLICT. Use ReplacingMergeTree for dedup.",
})
else:
@@ -319,7 +323,7 @@ class SQLGenerator:
dialect=dialect,
)
logger.reflect("SQL generated", {
log.reflect("SQL generated", payload={
"dialect": dialect,
"row_count": len(rows),
"sql_length": len(sql),

31
backend/src/plugins/translate/superset_executor.py
View File

@@ -18,9 +18,11 @@ import time
import uuid
from ...core.logger import logger, belief_scope
from ...core.cot_logger import MarkerLogger
from ...core.config_manager import ConfigManager
from ...core.superset_client import SupersetClient
log = MarkerLogger("SupersetExecutor")
# #region SupersetSqlLabExecutor [C:5] [TYPE Class] [SEMANTICS translate,superset,sqllab]
# @BRIEF Submit SQL to Superset SQL Lab API with polling and status tracking.
@@ -74,7 +76,7 @@ class SupersetSqlLabExecutor:
for db in databases:
if db.get("database_name", "").lower() == database_name.lower():
self._database_id = db["id"]
logger.reason("Resolved database ID by name", {
log.reason("Resolved database ID by name", payload={
"database_name": database_name,
"database_id": self._database_id,
})
@@ -85,7 +87,7 @@ class SupersetSqlLabExecutor:
# Default: use first database
self._database_id = databases[0]["id"]
logger.reason("Using default database", {
log.reason("Using default database", payload={
"database_name": databases[0].get("database_name"),
"database_id": self._database_id,
})
@@ -111,7 +113,7 @@ class SupersetSqlLabExecutor:
if db_uuid.lower() == uuid_str.lower():
db_id = db["id"]
self._database_id = db_id
logger.reason("Resolved database ID by UUID", {
log.reason("Resolved database ID by UUID", payload={
"uuid": uuid_str,
"database_id": db_id,
"database_name": db.get("database_name"),
@@ -145,7 +147,7 @@ class SupersetSqlLabExecutor:
except (ValueError, TypeError):
db_id = self.resolve_database_uuid(db_id)
logger.reason("Submitting SQL to Superset SQL Lab", {
log.reason("Submitting SQL to Superset SQL Lab", payload={
"database_id": db_id,
"sql_length": len(sql),
})
@@ -165,7 +167,7 @@ class SupersetSqlLabExecutor:
headers={"Content-Type": "application/json"},
)
except Exception as e:
logger.explore("SQL Lab execute failed", {"error": str(e)})
log.explore("SQL Lab execute failed", error=str(e))
raise ValueError(f"Superset SQL Lab execute failed: {e}")
# Parse response — try multiple formats
@@ -177,7 +179,7 @@ class SupersetSqlLabExecutor:
or (result.get("result") or {}).get("id"))
status = result.get("status", "unknown")
logger.reason("SQL Lab execute response", {
log.reason("SQL Lab execute response", payload={
"query_id": query_id,
"status": status,
"has_query_id": query_id is not None,
@@ -208,7 +210,7 @@ class SupersetSqlLabExecutor:
with belief_scope("SupersetSqlLabExecutor.poll_execution_status"):
client = self._get_client()
logger.reason("Polling SQL execution status", {
log.reason("Polling SQL execution status", payload={
"query_id": query_id,
"max_polls": max_polls,
"interval": poll_interval_seconds,
@@ -226,7 +228,7 @@ class SupersetSqlLabExecutor:
# Terminal states
if state in ("success", "finished", "completed"):
logger.reason("SQL execution completed", {
log.reason("SQL execution completed", payload={
"query_id": query_id,
"attempt": attempt + 1,
"rows_affected": result.get("rows"),
@@ -243,7 +245,8 @@ class SupersetSqlLabExecutor:
if state in ("failed", "error", "stopped"):
error_msg = result.get("error_message", "Unknown error")
logger.explore("SQL execution failed", {
log.explore("SQL execution failed", error="SQL query execution returned failed/error state",
payload={
"query_id": query_id,
"state": state,
"error": error_msg,
@@ -264,7 +267,8 @@ class SupersetSqlLabExecutor:
time.sleep(poll_interval_seconds)
except Exception as e:
logger.explore("Polling error, retrying", {
log.explore("Polling error, retrying", error="Polling encountered error, will retry",
payload={
"query_id": query_id,
"attempt": attempt + 1,
"error": str(e),
@@ -273,7 +277,7 @@ class SupersetSqlLabExecutor:
continue
# Timeout
logger.explore("SQL execution polling timed out", {
log.explore("SQL execution polling timed out", error="Polling timed out", payload={
"query_id": query_id,
"max_polls": max_polls,
})
@@ -305,7 +309,7 @@ class SupersetSqlLabExecutor:
query_id = exec_result.get("query_id")
if not query_id:
logger.explore("No query_id from SQL Lab execute", {
log.explore("No query_id from SQL Lab execute", error="No query_id returned", payload={
"raw_response": exec_result.get("raw_response"),
})
return {
@@ -341,7 +345,8 @@ class SupersetSqlLabExecutor:
"results": result.get("results"),
}
except Exception as e:
logger.explore("Failed to fetch query results", {
log.explore("Failed to fetch query results", error="Failed to fetch query results from Superset",
payload={
"query_id": query_id,
"error": str(e),
})