gitea-admin/cowork-local
2
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

refactor: xoá 1.400 dòng mã chết còn sót sau merge và 5 gói rỗng

Browse Source 
Hai bản tách song song của cùng một god-file cùng được giữ lại sau một lần
merge. Bản chết không ai import, và hai file trong đó còn không import nổi:
`graph_render.py` lấy `GraphQaMixin` không tồn tại, `task_actions.py` lấy
`ui.calendar_view` đã bị xoá.

Kèm theo 5 gói chỉ có `__init__.py` với docstring hứa những module chưa bao
giờ được tạo. Hai trong số đó (`adapters/qt/`, `infrastructure/platform/qt/`)
là vị trí đã bị bác bỏ có ghi lý do — `QtSchedulerClock` nằm ở
`infrastructure/qt/`, và lý do vì sao không đặt ở `platform/` vẫn còn nguyên
trong `infrastructure/qt/__init__.py`.

Không cổng nào bắt được đám này: file không ai import vẫn đúng chiều phụ
thuộc, vẫn sạch credential, vẫn dưới 400 dòng. Cổng O ở commit sau đi tìm
đúng khoảng trống đó.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
anhtnm1
2026-08-30 10:36:31 +09:00
co-authored by Claude Opus 5
parent 95b3b27578
commit a71085b39e
14 changed files with 0 additions and 1415 deletions
Download Patch File Download Diff File Expand all files Collapse all files
adapters/__init__.py
-12
View File
@@ -1,12 +0,0 @@
"""adapters/ — Adapter riêng cho Qt (clock, thread, timer).
Kế hoạch gốc đặt tên thư mục này là ``platform/``. Không dùng được: chạy
bất kỳ script nào từ thư mục gốc repo (``python tools/...``,
``python scripts/...``) thì ``platform/`` **che khuất module ``platform``
của thư viện chuẩn**, và ``import keyring`` chết ngay với
``AttributeError: module 'platform' has no attribute 'system'``.
Repo có 26 script chạy đúng kiểu đó.
Đổi tên là cách duy nhất chắc chắn — không thể bắt mọi người nhớ "đừng bao
giờ chạy python từ thư mục gốc".
"""
adapters/qt/__init__.py
View File
application/settings/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Application settings package: Settings application service."""
infrastructure/platform/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Infrastructure platform adapters package."""
infrastructure/platform/qt/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Infrastructure Qt platform adapters: QtSchedulerClock."""
presentation/folder/ai_edit_runner.py
-325
View File
@@ -1,325 +0,0 @@
"""Một lượt AI sửa file, từ lúc gửi tới lúc ghi ra đĩa — R08-T12.
``_ai_run_edit`` dài (82 dòng) vì nó là cả một lượt: dựng ngữ cảnh từ
file đang mở, gọi provider, nhận nội dung phát dần, tách phần mã khỏi
phần giải thích, rồi dựng bản xem trước.
Không bao giờ ghi đè thẳng: kết quả hiện ra để người dùng xem, và chỉ
``_ai_apply`` mới chạm vào file.
Cùng kiểu mixin, xem ghi chú ở đầu ``presentation/shell/nav_rail.py``.
"""
from __future__ import annotations
from .file_helpers import (
_HTML_SUFFIXES, _PPTX_SUFFIXES, _parse_ai_output, _pptx_available,
)
import os
from pathlib import Path
from typing import Optional
from PySide6.QtCore import Qt
from ...core.worker import AgentWorker
from ...i18n import tr
from ...theme import current_palette
class AiEditRunnerMixin:
"""Trộn vào FolderTab."""
def _ai_send(self) -> None:
if not self._root or not os.path.isdir(self._root):
self.ai_chat.add_error(tr("folder.ai_no_file"))
return
instruction = self.ai_input.text().strip()
if not instruction:
return
self.ai_input.clear()
self.ai_chat.add_user(instruction)
# QUEUE: while a run is active OR a proposal is awaiting Apply/Discard,
# hold the new instruction and run it when the pipeline goes idle. Lets
# the user line up several edits without waiting for each to finish.
if self._ai_worker is not None or self._ai_pending is not None:
self._ai_queue.append(instruction)
self.ai_chat.add_status(tr("folder.ai_queued", n=len(self._ai_queue)))
self._update_queue_status()
return
self._ai_start(instruction)
def _ai_start(self, instruction: str) -> None:
"""Begin processing one instruction (plan → edit). Assumes the pipeline
is idle (the queue calls this when the previous run finishes)."""
# If a text/code/HTML file is open (even in Preview), switch it into the
# editor so AI can edit it. If nothing editable is open, that's fine —
# the request may be to CREATE a new file (the model names it via FILE:).
editable = self.stack.currentWidget() is self.editor
if not editable:
editable = self._ensure_editor_for_ai()
self._maybe_suggest_image_model(instruction)
# Auto Model Routing (may switch to the best coding model for this run).
self._ai_apply_routing(instruction)
has_file = editable and bool(self._current_file)
self._ai_running_file = Path(self._current_file).name if has_file else tr("folder.ai_new_file")
self._ai_set_busy(True)
# Announce start on the status bar so it's visible even from another tab —
# the edit keeps running in the background until it finishes.
self.status_message.emit(tr("folder.ai_running", name=self._ai_running_file))
# Two phases so the PLAN is shown INLINE *before* the edit runs.
self._ai_ctx = {
"filename": Path(self._current_file).name if has_file else "",
"content": self.editor.toPlainText() if has_file else "",
"convo": self._cowork_context(),
"instruction": instruction,
"provider": self._ai_provider(),
"plan": "",
}
# Reset the token/cost tally for THIS prompt (plan + edit calls sum into it).
self._ai_prompt_usage = {"in": 0, "out": 0, "cache": 0, "cost": 0.0}
self._ai_run_plan()
def _ai_maybe_dequeue(self) -> None:
"""When the pipeline is fully idle, start the next queued instruction."""
if self._ai_worker is not None or self._ai_pending is not None:
return
if not self._ai_queue:
return
nxt = self._ai_queue.pop(0)
self._update_queue_status()
self._ai_start(nxt)
def _ai_add_usage(self, usage) -> None:
"""Add one model call's usage (plan or edit) to THIS prompt's tally."""
if not isinstance(usage, dict):
return
tot = getattr(self, "_ai_prompt_usage", None)
if tot is None:
tot = self._ai_prompt_usage = {"in": 0, "out": 0, "cache": 0, "cost": 0.0}
tot["in"] += int(usage.get("in", 0) or 0)
tot["out"] += int(usage.get("out", 0) or 0)
tot["cache"] += int(usage.get("cache", 0) or 0)
tot["cost"] += float(usage.get("cost_usd", 0.0) or 0.0)
def _ai_show_usage(self, bubble) -> None:
"""Footer under the AI-edit reply: ↓in ↑out ▤ctx $cost for the whole
prompt (plan + edit), priced in the display currency — same as Cowork."""
tot = getattr(self, "_ai_prompt_usage", None)
if bubble is None or not tot or not (tot["in"] or tot["out"]):
return
from ...core import model_pricing as mp, usage_tracker as ut
pricing = {**ut.DEFAULT_PRICING, **(self.ctx.config.data.get("usage") or {})}
line = (f"↓{mp.format_tokens(int(tot['in']))} ↑{mp.format_tokens(int(tot['out']))} "
f"▤{mp.format_tokens(int(tot['in'] + tot['out'] + tot['cache']))} "
f"{ut.format_cost(tot['cost'], pricing)}")
try:
bubble.add_usage(line)
except Exception: # noqa: BLE001 - a usage footer must never break the edit
pass
def _ai_run_plan(self) -> None:
c = self._ai_ctx
plan_bubble = self.ai_chat.add_plan(tr("folder.ai_planning"))
self.ai_chat.scroll_to_bottom()
def job(worker):
from ...core import usage_tracker as ut
from ...core.co4e_runner import _usage_delta
provider = c["provider"]
messages = [{"role": "system", "content":
"You are an AI file editor. Give a SHORT numbered plan (2-4 steps) for "
"the requested change. Plan ONLY — do NOT output any code."}]
if c["convo"]:
messages.append({"role": "system",
"content": "Context from the user's Cowork conversation:\n" + c["convo"]})
messages.append({"role": "user", "content":
f"File: {c['filename']}\n\nCurrent content:\n```\n{c['content']}\n```\n\n"
f"Request: {c['instruction']}"})
ut.set_context("folder", c.get("filename") or "AI edit") # Dashboard + usage
ut.begin_accumulation(); base = ut.accumulated()
try:
r = provider.chat(messages, tools=None, cancel=worker.is_cancelled)
txt = r.get("content", "") if isinstance(r, dict) else str(r)
usage = _usage_delta(base, self.ctx.config)
finally:
ut.end_accumulation()
return {"plan": provider.strip_think(txt) or "", "usage": usage}
worker = AgentWorker(job)
worker.finished_ok.connect(lambda res, b=plan_bubble: self._ai_plan_done(res, b))
worker.failed.connect(lambda err, b=plan_bubble: self._ai_failed(err, b))
self._ai_worker = worker
worker.start()
def _ai_plan_done(self, result, plan_bubble) -> None:
self._ai_add_usage((result or {}).get("usage")) # plan-step tokens
plan = ((result or {}).get("plan") or "").strip()
self._ai_ctx["plan"] = plan
plan_bubble.set_plain(plan or tr("folder.ai_empty"))
self.ai_chat.scroll_to_bottom()
self._ai_run_edit() # now execute the plan
def _ai_run_edit(self) -> None:
c = self._ai_ctx
bubble = self.ai_chat.add_assistant(tr("folder.ai_edit"))
self.ai_chat.scroll_to_bottom()
pptx_note = ("\nThis is a PPTX shown as marker blocks '### Slide N / Box M', where N is the "
"1-based SLIDE NUMBER and M the box on that slide. When the user refers to a "
"slide by number (e.g. 'edit slide 3'), change ONLY the blocks under '### Slide "
"3' and leave every other slide's block exactly as-is. Each block has fields "
"type/pos/size/font/text (and image for pictures). To change TEXT COLOUR or "
"FONT edit the `font:` line — e.g. `font: name=Arial size=28 bold=1 "
"color=FF0000`. Keep all block markers and structure.") if self._edit_kind == "pptx" else ""
# When creating a NEW deck (request mentions slides/pptx and we're not
# already editing one), tell the model the marker format to emit so we can
# build a real .pptx from it.
_pptx_words = ("pptx", "powerpoint", "slide", "presentation", "deck",
"スライド", "プレゼン", "trình chiếu", "trinh chieu", "bài thuyết trình")
wants_new_pptx = (self._edit_kind != "pptx"
and any(w in c["instruction"].lower() for w in _pptx_words))
new_pptx_note = ("\nTo CREATE a PowerPoint, name it `FILE: <name>.pptx` and output the slides "
"as marker blocks — one block per shape:\n"
"### Slide 1 / Box 1\ntype: text\npos: 0.5, 0.4\nsize: 9.0, 1.2\n"
"font: name=Calibri size=32 bold=1 color=1F3864\ntext:\nTitle here\n\n"
"### Slide 1 / Box 2\ntype: text\npos: 0.5, 1.8\nsize: 9.0, 4.5\n"
"text:\nBullet one\nBullet two\n\n"
"Increment the Slide number for each new slide; pos/size are in inches; "
"font color is RRGGBB hex.") if wants_new_pptx else ""
imggen_note = ""
try:
from ...core import image_gen
if image_gen.is_configured(self.ctx.config):
imggen_note = ("\nYou can also GENERATE an illustration image: add a line "
"`IMAGE_GEN: <describe the image> => <relative/path.png>`. Use a "
"generated image e.g. as a new picture, or (for pptx) set a picture "
"box's `image:` field to that same path to insert it.")
except Exception: # noqa: BLE001
pass
def job(worker):
provider = c["provider"]
open_note = (f"the currently-open file '{c['filename']}'" if c["filename"]
else "no file is open")
messages = [{"role": "system", "content":
"You are an AI file editor inside an app. Following the plan, output the "
"COMPLETE file content in ONE fenced code block (```), and nothing after "
"it. Preserve everything you were not asked to change.\n"
"If the request is to CREATE A NEW file (or a different file than the one "
"open), put a line `FILE: <relative/path/name.ext>` (relative to the "
"current folder) immediately before the code block. Omit FILE to edit the "
f"open file. Right now, {open_note}." + pptx_note + new_pptx_note + imggen_note}]
if c["convo"]:
messages.append({"role": "system",
"content": "Context from the user's Cowork conversation:\n" + c["convo"]})
if c["plan"]:
messages.append({"role": "system", "content": "Plan to follow:\n" + c["plan"]})
cur = (f"File: {c['filename']}\n\nCurrent content:\n```\n{c['content']}\n```\n\n"
if c["filename"] else "No file is currently open.\n\n")
messages.append({"role": "user", "content": cur + f"Request: {c['instruction']}"})
def on_text(piece: str) -> None:
worker.emit_event({"type": "text", "delta": piece})
from ...core import usage_tracker as ut
from ...core.co4e_runner import _usage_delta
ut.set_context("folder", c.get("filename") or "AI edit") # Dashboard + usage
ut.begin_accumulation(); base = ut.accumulated()
try:
r = provider.chat(messages, tools=None, on_text=on_text, cancel=worker.is_cancelled)
txt = r.get("content", "") if isinstance(r, dict) else str(r)
usage = _usage_delta(base, self.ctx.config)
finally:
ut.end_accumulation()
return {"text": provider.strip_think(txt) or "", "usage": usage}
worker = AgentWorker(job)
worker.event.connect(lambda ev, b=bubble: self._ai_stream(ev, b))
worker.finished_ok.connect(lambda res, b=bubble: self._ai_done(res, b))
worker.failed.connect(lambda err, b=bubble: self._ai_failed(err, b))
self._ai_worker = worker
worker.start()
def _ai_stream(self, ev, bubble) -> None:
if isinstance(ev, dict) and ev.get("type") == "text":
bubble.append_delta(ev.get("delta", ""))
self.ai_chat.scroll_to_bottom()
def _ai_done(self, result, bubble) -> None:
self._ai_worker = None
self._ai_set_busy(False)
self._ai_add_usage((result or {}).get("usage")) # edit-step tokens
self._ai_show_usage(bubble) # footer: prompt total (plan+edit)
text = ((result or {}).get("text") or "").strip()
target, new_content, summary, image_gens = _parse_ai_output(text)
if new_content is None and not image_gens:
bubble.set_markdown(text or tr("folder.ai_empty"))
self.ai_chat.scroll_to_bottom()
self._ai_flag_done()
return
# Decide edit-current vs create-new. A FILE: naming a path different from
# the open file (or when nothing is open) → CREATE a new file.
create = bool(target) and (not self._current_file
or Path(target).name != Path(self._current_file).name)
# PROPOSE the change — nothing is written until the user clicks Apply.
self._ai_pending = {"content": new_content,
"target": target if create else None,
"image_gens": image_gens}
hint = tr("folder.ai_review_hint")
bubble.set_markdown((summary + "\n\n" if summary else "") + "_" + hint + "_")
if new_content is not None:
import difflib
old = "" if create else self.editor.toPlainText()
diff = "".join(difflib.unified_diff(
old.splitlines(keepends=True), new_content.splitlines(keepends=True),
fromfile=("(new file)" if create else "current"),
tofile=(target if create else "proposed"))) or "(no textual difference)"
title = tr("folder.ai_proposed_new", name=target) if create else tr("folder.ai_proposed")
self.ai_chat.add_diff(title, diff)
if image_gens:
listing = "\n".join(f"• {p} ({prompt[:60]})" for prompt, p in image_gens)
self.ai_chat.add_status(tr("folder.ai_image_plan") + "\n" + listing)
self._ai_confirm_row.setVisible(True)
self.ai_chat.scroll_to_bottom()
name = target if create else getattr(self, "_ai_running_file", "")
self.status_message.emit(tr("folder.ai_proposed_status", name=name))
self._ai_status.setText("● " + hint)
self._ai_status.setStyleSheet(f"color:{current_palette().warning};")
def _ai_apply(self) -> None:
"""Confirmed by the user. If the edit GENERATES images, ask the image
gate then generate them (off-thread) before finalising the file edit."""
if not self._ai_pending:
return
p = self._ai_pending
self._ai_pending = None
self._ai_confirm_row.setVisible(False)
if p.get("image_gens"):
from PySide6.QtWidgets import QMessageBox
if QMessageBox.question(self, tr("folder.ai_image_confirm_title"),
tr("folder.ai_image_confirm_gen")) != QMessageBox.Yes:
self.status_message.emit(tr("folder.ai_image_declined"))
return
self._ai_generate_then_finalize(p)
return
self._ai_finalize_apply(p)
def _ai_discard(self) -> None:
self._ai_pending = None
self._ai_confirm_row.setVisible(False)
self.ai_chat.add_status(tr("folder.ai_discarded"))
self.ai_chat.scroll_to_bottom()
self._ai_status.setText("")
self._ai_maybe_dequeue() # discarding resolves the gate → run the next queued edit
def _ai_failed(self, err, bubble) -> None:
self._ai_worker = None
bubble.set_markdown(tr("folder.ai_error", err=err))
self._ai_set_busy(False)
self.status_message.emit(tr("folder.ai_error", err=err))
self._ai_flag_done()
presentation/folder/ai_output_writer.py
-140
View File
@@ -1,140 +0,0 @@
"""Ghi kết quả AI ra đĩa — R08-T12.
Tách khỏi ``ai_edit_runner.py`` vì đây là phần DUY NHẤT thật sự chạm vào
file của người dùng. Mọi thứ trước nó chỉ dựng bản xem trước.
Gồm cả nhánh sinh ảnh: lượt nào có ảnh thì phải chờ ảnh xong mới ghi, vì
nội dung có thể tham chiếu tới đường dẫn ảnh vừa tạo.
"""
from __future__ import annotations
from .file_helpers import (
_HTML_SUFFIXES, _PPTX_SUFFIXES, _parse_ai_output, _pptx_available,
)
import os
from pathlib import Path
from typing import Optional
from PySide6.QtCore import Qt
from ...core.worker import AgentWorker
from ...i18n import tr
from ...theme import current_palette
class AiOutputWriterMixin:
"""Trộn vào FolderTab."""
def _ai_generate_then_finalize(self, p: dict) -> None:
imgs = p.get("image_gens") or []
root = os.path.normpath(self._root)
img_model, img_base, img_key = self._ai_image_model() # may target another provider
self._ai_set_busy(True)
self.status_message.emit(tr("folder.ai_generating"))
def job(worker):
from ...core import image_gen
results = []
for prompt, rel in imgs:
dest = rel if os.path.isabs(rel) else os.path.join(root, rel)
dest = os.path.normpath(dest)
if os.path.commonpath([dest, root]) != root:
results.append((rel, False, "path escapes the folder"))
continue
try:
os.makedirs(os.path.dirname(dest) or root, exist_ok=True)
except OSError as exc:
results.append((rel, False, str(exc)))
continue
ok, msg = image_gen.generate_image(self.ctx.config, prompt, dest,
model=img_model, base_url=img_base, api_key=img_key)
results.append((dest, ok, msg))
return {"results": results}
worker = AgentWorker(job)
worker.finished_ok.connect(lambda res, pp=p: self._ai_images_done(res, pp))
worker.failed.connect(lambda err, pp=p: self._ai_images_done({"results": [], "err": err}, pp))
self._ai_worker = worker
worker.start()
def _ai_images_done(self, res: dict, p: dict) -> None:
self._ai_worker = None
self._ai_set_busy(False)
created = []
for dest, ok, msg in res.get("results", []):
if ok:
created.append(dest)
self.ai_chat.add_success("✓ " + tr("folder.ai_image_created", name=Path(dest).name))
else:
self.ai_chat.add_error(tr("folder.ai_image_failed", err=msg))
# Now apply any text/file edit (pptx image: fields now point at real files).
self._ai_finalize_apply(p, images_done=True)
# If it was only image generation, open the first new image.
if p.get("content") is None and not p.get("target") and created:
self.open_file(created[0], reset=False)
def _ai_finalize_apply(self, p: dict, images_done: bool = False) -> None:
content = p.get("content")
target = p.get("target")
if content is None:
self.ai_chat.scroll_to_bottom()
self._ai_flag_done()
self.status_message.emit(tr("folder.ai_done", name=getattr(self, "_ai_running_file", "")))
return
if target:
dest = self._create_new_file(target, content)
if dest is None:
return
self.ai_chat.add_success("✓ " + tr("folder.ai_created", name=Path(dest).name))
self.status_message.emit(tr("folder.ai_created", name=Path(dest).name))
else:
self.editor.setPlainText(content) # live update in the editor/preview
self._ai_write_out(content, skip_image_confirm=images_done)
self.ai_chat.add_success("✓ " + tr("folder.ai_applied"))
self.status_message.emit(tr("folder.ai_done", name=getattr(self, "_ai_running_file", "")))
self.ai_chat.scroll_to_bottom()
self._ai_flag_done()
def _create_new_file(self, target: str, content: str) -> Optional[str]:
"""Create ``target`` (relative to the folder root) with ``content`` and
open it — like Cowork's save_file. Refuses paths escaping the root."""
root = os.path.normpath(self._root)
dest = target if os.path.isabs(target) else os.path.join(root, target)
dest = os.path.normpath(dest)
if os.path.commonpath([dest, root]) != root:
self.status_message.emit(tr("folder.ai_error", err="path escapes the folder"))
return None
try:
os.makedirs(os.path.dirname(dest) or root, exist_ok=True)
if Path(dest).suffix.lower() in _PPTX_SUFFIXES and _pptx_available():
# A .pptx is a binary package — build a real deck from the marker
# text (writing text straight to .pptx would corrupt it).
from ...core import pptx_edit
pptx_edit.create_pptx_from_text(dest, content)
else:
Path(dest).write_text(content, encoding="utf-8")
except Exception as exc: # noqa: BLE001 - OS error or pptx build failure
self.status_message.emit(tr("folder.save_error", err=str(exc)))
return None
self.open_file(dest, reset=False) # show the new file; keep this AI chat
return dest
def _ai_write_out(self, content: str, skip_image_confirm: bool = False) -> None:
"""Persist the confirmed content to disk AND refresh the preview.
pptx text is written back into the deck (no PowerPoint window)."""
if not self._current_file:
return
try:
if self._edit_kind == "pptx":
if not self._write_pptx(content, skip_confirm=skip_image_confirm):
return
else:
Path(self._current_file).write_text(content, encoding="utf-8")
except Exception as exc: # noqa: BLE001
self.status_message.emit(tr("folder.save_error", err=str(exc)))
return
# Refresh preview: HTML re-renders; pptx re-renders the slides; code stays
# in the (now-saved) editor.
suffix = Path(self._current_file).suffix.lower()
if suffix in _HTML_SUFFIXES:
self._show_html(self._current_file, mode_preview=True)
elif suffix in _PPTX_SUFFIXES:
self._show_pptx(self._current_file, mode_preview=True)
presentation/folder/file_helpers.py
-114
View File
@@ -1,114 +0,0 @@
"""Hàm phụ trợ đọc và nhận dạng file — R08-T12.
Thuần hàm, không widget. ``_is_probably_text`` là chỗ quyết định file
mở bằng ô soạn thảo hay báo là nhị phân — đoán sai thì người dùng thấy
một màn hình ký tự rác.
"""
from __future__ import annotations
from ...ui.libreoffice_view import DOC_SUFFIXES # noqa: F401 — dùng lại ở cả gói
# Nhận dạng loại file và các ngưỡng — gom về đây vì cả sáu file trong gói
# đều hỏi tới, để rải ra thì mỗi lần thêm một đuôi file phải sửa vài chỗ.
try:
from ...graph.graph_web import _HAS_WEB
except Exception: # pragma: no cover
_HAS_WEB = False
try:
from PySide6.QtPdf import QPdfDocument # noqa: F401
from PySide6.QtPdfWidgets import QPdfView # noqa: F401
_HAS_PDF = True
except Exception: # pragma: no cover - QtPdf not bundled
_HAS_PDF = False
_IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp", ".ico"}
_HTML_SUFFIXES = {".html", ".htm"}
_PPTX_SUFFIXES = {".pptx"} # text-editable in-app via python-pptx (no PowerPoint)
_EXCEL_SUFFIXES = {".xlsx", ".xlsm"} # viewed as a real table via openpyxl (no LibreOffice)
_MAX_EDIT_BYTES = 2_000_000 # above this, view read-only / note only
_MAX_HIGHLIGHT_CHARS = 300_000 # skip colouring huge files (keeps typing snappy)
import os
from pathlib import Path
from PySide6.QtCore import Qt
from PySide6.QtGui import QColor, QFont, QTextCharFormat
from ...i18n import tr
def _fmt(color: str, *, italic: bool = False, bold: bool = False) -> QTextCharFormat:
f = QTextCharFormat()
f.setForeground(QColor(color))
if italic:
f.setFontItalic(True)
if bold:
f.setFontWeight(QFont.Bold)
return f
def _pptx_available() -> bool:
"""True when python-pptx is importable. If it's MISSING, auto-download &
install it (via deps.ensure_module) so pptx editing 'just works' — cached so
the (one-time) install is attempted only once."""
global _PPTX_READY
if _PPTX_READY is None:
try:
from ...core.deps import ensure_module
_PPTX_READY = ensure_module("pptx", "python-pptx") is not None
except Exception: # noqa: BLE001
_PPTX_READY = False
return _PPTX_READY
def _split_code_block(text: str):
"""Split an AI reply into ``(file_content, summary)``. ``file_content`` is
the first fenced code block (the edited file); ``summary`` is any prose
before it. Returns ``(None, text)`` when there's no code block."""
import re
m = re.search(r"```[^\n]*\n(.*?)```", text or "", re.DOTALL)
if not m:
return None, (text or "")
return m.group(1), (text[:m.start()].strip())
def _parse_ai_output(text: str):
"""Parse an AI edit reply into ``(target, content, summary, image_gens)``.
``FILE: <path>`` names a NEW file to create; ``IMAGE_GEN: <prompt> => <path>``
lines request generated illustration images (relative paths)."""
import re
content, summary = _split_code_block(text)
target = None
m = re.search(r"(?mi)^\s*FILE:\s*(.+?)\s*$", text or "")
if m:
target = m.group(1).strip().strip("`\"'")
image_gens = []
for gm in re.finditer(r"(?mi)^\s*IMAGE_GEN:\s*(.+?)\s*=>\s*(\S+)\s*$", text or ""):
image_gens.append((gm.group(1).strip(), gm.group(2).strip().strip("`\"'")))
# Strip the directive lines out of the shown summary.
summary = re.sub(r"(?mi)^\s*(FILE|IMAGE_GEN):\s*.+?$", "", summary).strip()
return target, content, summary, image_gens
def _read_text(path: str) -> str:
try:
return Path(path).read_text(encoding="utf-8", errors="replace")
except OSError as exc:
return f"[could not read file: {exc}]"
def _is_probably_text(path: str) -> bool:
try:
with open(path, "rb") as f:
chunk = f.read(4096)
except OSError:
return False
if b"\x00" in chunk:
return False
try:
chunk.decode("utf-8")
return True
except UnicodeDecodeError:
# Latin-ish text still edits fine via errors="replace"; only reject on
# a hard binary signal (NUL above), so most source files pass.
return True
presentation/folder/image_model_picker.py
-115
View File
@@ -1,115 +0,0 @@
"""Chọn model sinh ảnh cho AI sửa file — R08-T12.
Dò khắp mọi provider đã cấu hình xem cái nào sinh được ảnh, rồi gợi ý khi
câu người dùng gõ nghe như đang muốn tạo ảnh. Có thể gợi ý model của
provider KHÁC provider đang chọn — nên nó tách riêng: đây là chỗ duy nhất
trong màn Thư mục biết tới nhiều provider cùng lúc.
"""
from __future__ import annotations
from .file_helpers import (
DOC_SUFFIXES, _HTML_SUFFIXES, _IMAGE_SUFFIXES, _PPTX_SUFFIXES, _is_probably_text, _pptx_available,
)
import os
from pathlib import Path
from PySide6.QtCore import Qt, Signal
from PySide6.QtWidgets import QComboBox, QHBoxLayout, QLabel, QLineEdit, QPushButton, QVBoxLayout, QWidget
from ...core.worker import AgentWorker
from ...i18n import tr
from ...theme import current_palette
from ...ui.chat_view import ChatView
from ...ui.libreoffice_view import DOC_SUFFIXES
class ImageModelPickerMixin:
"""Trộn vào FolderTab."""
def _scan_all_image_models(self, then_suggest: bool = False) -> None:
"""Background: find image-capable models across EVERY configured provider
(not just the active one), so we can suggest one when an edit involves
images even if the active provider has none. Caches
``self._all_image_models = [(provider_key, model)]``."""
if self._img_scan_worker is not None:
if then_suggest:
self._pending_img_suggest = True
return
providers = dict(self.ctx.config.data.get("providers", {}))
# Only providers that actually have an endpoint/key configured.
candidates = [k for k, c in providers.items()
if (c.get("base_url") or c.get("api_key"))]
def job(worker):
from ...core import image_gen
found = []
for key in candidates:
try:
prov = self.ctx.build_provider_for(key)
models = list(getattr(prov, "list_models", lambda: [])() or [])
except Exception: # noqa: BLE001 - a broken provider must not block the scan
models = []
for m in models:
if image_gen.looks_like_image_model(m):
found.append((key, m))
return {"found": found}
def done(res):
self._img_scan_worker = None
self._all_image_models = list(res.get("found", []))
if getattr(self, "_pending_img_suggest", False):
self._pending_img_suggest = False
self._suggest_cross_provider_image()
w = AgentWorker(job)
w.finished_ok.connect(done)
w.failed.connect(lambda _e: setattr(self, "_img_scan_worker", None))
self._img_scan_worker = w
if then_suggest:
self._pending_img_suggest = True
w.start()
def _maybe_suggest_image_model(self, instruction: str) -> None:
"""If the request looks image-related, suggest a suitable image model
BEFORE running — searching the active provider first, then ALL providers.
The suggested model is what image generation will auto-use."""
from ...core import image_gen
low = (instruction or "").lower()
if not any(w in low for w in self._IMAGE_WORDS):
return
picked = self.ai_model_combo.currentData()
if picked and image_gen.looks_like_image_model(picked):
return
local = image_gen.suggest_image_model(self._ai_models)
if local:
self.ai_chat.add_status(tr("folder.ai_image_suggest", model=local))
return
# None on the active provider → look across ALL providers (cached, or scan
# now and suggest when the scan returns).
if self._all_image_models:
self._suggest_cross_provider_image()
elif self._img_scan_worker is not None:
self._pending_img_suggest = True # a scan is already running
else:
self._scan_all_image_models(then_suggest=True)
def _suggest_cross_provider_image(self) -> None:
"""Post a suggestion listing image models found on OTHER providers. When
none exist anywhere, fall back to telling the user their PICKED model
will be used for image generation (or that there's nothing to use)."""
from ...config import PROVIDER_LABELS
if not self._all_image_models:
picked = self.ai_model_combo.currentData()
if picked:
self.ai_chat.add_status(tr("folder.ai_image_use_selected", model=picked))
else:
self.ai_chat.add_status(tr("folder.ai_image_none"))
return
seen, lines = set(), []
for key, model in self._all_image_models:
tag = (key, model)
if tag in seen:
continue
seen.add(tag)
lines.append(f"• {model} ({PROVIDER_LABELS.get(key, key)})")
if len(lines) >= 5:
break
self.ai_chat.add_status(tr("folder.ai_image_suggest_all") + "\n" + "\n".join(lines))
presentation/graph/graph_project.py
-109
View File
@@ -1,109 +0,0 @@
"""Chọn project và đổi chế độ xem cho GraphRAG — R08-T14.
Đồ thị luôn thuộc về một project. Đổi project là phải quét lại từ đầu, nên
phần này giữ luôn việc dọn kết quả cũ trước khi nạp cái mới.
Cùng kiểu mixin, xem ghi chú ở đầu ``presentation/shell/nav_rail.py``.
"""
from __future__ import annotations
from .graph_qa_widget import GraphQaMixin
from .graph_render import GraphRenderMixin
from .graph_scene import _Edge, _GraphView, _Node
import re
import sys
from pathlib import Path
from PySide6.QtCore import QPointF, Qt, QTimer, Signal
from PySide6.QtGui import QColor
from PySide6.QtWidgets import QComboBox, QFileDialog, QGraphicsScene, QGraphicsView, QHBoxLayout, QLabel, QLineEdit, QPushButton, QSplitter, QStackedWidget, QTabBar, QTextBrowser, QVBoxLayout, QWidget
from ...theme import current_palette
from ...core.worker import AgentWorker
from ...i18n import on_language_changed, tr
from ...state import AppContext
from ...ui.icons import collapse_right_icon, icon
from ...ui.widgets import CollapseStrip
class GraphProjectMixin:
"""Chọn project + đổi tab xem. Trộn vào StructureGraphView."""
def _retranslate(self) -> None:
self.path_edit.setPlaceholderText(tr("structure.path_placeholder"))
self._pick_btn.setText(tr("structure.browse"))
self._scan_btn.setText(tr("structure.scan"))
self._export_btn.setText(tr("structure.export_png"))
# Both views are named at once now, so neither label depends on state.
self.view_tabs.setTabText(0, tr("structure.graph_btn"))
self.view_tabs.setTabText(1, tr("structure.msgs_btn"))
self.view_tabs.setTabToolTip(1, tr("structure.msgs_tooltip"))
self._ag_collapse.setToolTip(tr("structure.collapse_agent_tooltip"))
self._ag_label.setText(tr("structure.agent_header"))
self.ask_edit.setPlaceholderText(tr("structure.ask_placeholder"))
self._ask_btn.setText(tr("structure.ask"))
if self._detail_mode == "idle":
self.detail.setPlaceholderText(tr("structure.detail_placeholder"))
self._agent_strip.setToolTip(tr("structure.expand_agent_tooltip"))
self.project_combo.setToolTip(tr("structure.project_tooltip"))
self._refresh_project_combo()
def _refresh_project_combo(self) -> None:
from ...core.projects import list_projects
keep = self._active_project_id
self.project_combo.blockSignals(True)
self.project_combo.clear()
self.project_combo.addItem(tr("structure.project_none"), "")
row_to_select = 0
for i, p in enumerate(list_projects(), start=1):
self.project_combo.addItem(p.name, p.project_id)
if p.project_id == keep:
row_to_select = i
self.project_combo.setCurrentIndex(row_to_select)
self.project_combo.blockSignals(False)
def set_project(self, project_id: str) -> None:
pid = project_id or ""
self._refresh_project_combo()
target = self.project_combo.findData(pid)
if target < 0:
target = 0
if self.project_combo.currentIndex() == target:
self._on_project_changed(target)
else:
self.project_combo.setCurrentIndex(target)
def _on_project_changed(self, _idx: int) -> None:
from ...core.projects import load_project
pid = self.project_combo.currentData() or ""
project_changed = pid != self._active_project_id
if project_changed:
self._clear_extracts() # different workspace → drop temp extraction
self._active_project_id = pid
locked = bool(pid)
self.path_edit.setReadOnly(locked)
# Also disable the folder-pick button — otherwise the scan path is only
# "locked" against typing, but the picker could still repoint it outside
# the selected project's sandbox, breaking GraphRAG scope isolation.
self._pick_btn.setEnabled(not locked)
if locked:
project = load_project(pid)
if project is not None:
self.path_edit.setText(str(project.workspace_dir()))
if project_changed:
# Mark it and scan on the next visit rather than now. The rail's
# project picker made switching a one-click thing from any screen,
# and each switch rebuilt this graph — a folder walk plus a force
# layout plus a full setHtml of the D3 page — for a tab that was
# usually not even on screen. auto_scan_and_fit() picks the flag up
# when GraphRAG is actually opened.
self._needs_scan = True
def _pick(self) -> None:
chosen = QFileDialog.getExistingDirectory(self, tr("structure.pick_folder_title"), self.path_edit.text())
if chosen:
self.path_edit.setText(chosen)
def _on_view_tab(self, index: int) -> None:
"""Tab 0 = graph, tab 1 = messages. Same two views as before, now named
on screen instead of hidden behind one button's changing label."""
if index == 1:
self._reload_messages()
self._stack.setCurrentWidget(self._msgs_view)
else:
self._stack.setCurrentWidget(self.web if self.web is not None else self.view)
presentation/graph/graph_render.py
-227
View File
@@ -1,227 +0,0 @@
"""Quét mã nguồn, dựng đồ thị, và xuất ảnh — R08-T14.
Hai đường vẽ song song: khung nhìn Qt (``_render``) và bản D3 chạy trong
QtWebEngine (``_render_d3``). WebEngine nặng nên chỉ dựng ở lần hiện đầu tiên
(``_ensure_web``), và ``prewarm`` hâm nóng nó lúc rảnh để bấm vào GraphRAG
không phải ngồi nhìn khung trắng.
Cùng kiểu mixin, xem ghi chú ở đầu ``presentation/shell/nav_rail.py``.
"""
from __future__ import annotations
from .graph_scene import _Bridge, _Edge, _Node
from .graph_web import _HAS_WEB, QWebChannel, QWebEngineView
import math
import re
from pathlib import Path
from PySide6.QtCore import QPointF, Qt, QUrl
from PySide6.QtGui import QColor
from PySide6.QtWidgets import QFileDialog
from ...theme import current_palette
from ...core.worker import AgentWorker
from ...i18n import tr
class GraphRenderMixin:
"""Quét, vẽ, xuất. Trộn vào StructureGraphView."""
def schedule_rescan(self, path: str = "") -> None:
if self._graph is None:
self._needs_scan = True
return
self._rescan_timer.start()
def prewarm(self) -> None:
"""Pay for the graph view before it is clicked on, not during.
Opening GraphRAG built a QWebEngineView (~140ms) and scanned the project
(~485ms) while an empty browser sat on screen — long enough, and white
enough, to read as the app restarting itself. Called from an idle timer
after the window is up, so startup itself is unaffected; the memory the
lazy construction was saving is spent a few seconds later instead.
"""
if not _HAS_WEB or self.web is not None:
return
self._ensure_web()
if self._graph is None and self.path_edit.text().strip():
self._needs_scan = False
self._scan() # runs on a worker thread
def _ensure_web(self) -> None:
if self.web is not None or not _HAS_WEB:
return
self.web = QWebEngineView()
# Blank the page in the app's own background first. A fresh
# QWebEngineView paints white, and on a dark theme that white rectangle
# WAS the flash — it showed for as long as the first scan took.
self.web.setHtml(
f"<body style='margin:0;background:{current_palette().bg}'></body>")
self._bridge = _Bridge()
self._channel = QWebChannel()
self._channel.registerObject("py", self._bridge)
self.web.page().setWebChannel(self._channel)
self._stack.addWidget(self.web)
self._stack.setCurrentWidget(self.web)
if self._graph is not None:
self._render_d3()
def auto_scan_and_fit(self) -> None:
self._ensure_web()
if not self.path_edit.text().strip():
return
if getattr(self, "_worker", None) is not None and self._worker.isRunning():
self._fit()
self._preserve_answer()
return
if self._graph is not None and not self._needs_scan:
self._fit()
self._preserve_answer()
return
self._needs_scan = False
self._scan()
def _scan(self) -> None:
path = self.path_edit.text().strip() or str(Path.cwd())
mode = "files" # default: scan all files (filter removed)
use_cmem = bool(self.ctx.config.codebase_memory.get("enabled"))
cmem_bin = self.ctx.config.codebase_memory.get("binary_path", "")
st = self.ctx.config.structure
max_nodes = int(st.get("max_nodes", 500) or 0)
max_edges = int(st.get("max_edges", 500) or 0)
self._scan_seq += 1
seq = self._scan_seq
self.status_message.emit(tr("structure.scanning"))
def job(worker: AgentWorker):
from ...core.structure_graph import (
build_from_codebase_memory, build_from_directory, force_layout,
)
if use_cmem:
from ...core.codebase_memory import CodebaseMemory
mem = CodebaseMemory(cmem_bin)
graph = (build_from_codebase_memory(mem, path, mode, max_nodes, max_edges)
if mem.available else build_from_directory(path, mode, max_nodes, max_edges))
else:
graph = build_from_directory(path, mode, max_nodes, max_edges)
pos = force_layout(graph)
return {"graph": graph, "pos": pos, "seq": seq}
w = AgentWorker(job)
w.finished_ok.connect(self._render)
w.failed.connect(lambda e: self.status_message.emit(tr("structure.scan_error", err=e)))
self._worker = w
w.start()
def _render(self, result: dict) -> None:
if result.get("seq") is not None and result["seq"] != self._scan_seq:
return
graph = result.get("graph")
pos = result.get("pos", {})
if graph is None:
return
self._graph = graph
self.scene.clear()
self.scene.setBackgroundBrush(QColor(current_palette().bg)) # restore after clear
self._node_items = []
self._edge_items = []
degree = {n.id: 0 for n in graph.nodes}
for e in graph.edges:
if e.source in degree:
degree[e.source] += 1
if e.target in degree:
degree[e.target] += 1
items = {}
sx = sy = 0.0
for node in graph.nodes:
radius = int(8 + min(20, 2.2 * math.sqrt(degree.get(node.id, 0))))
item = _Node(node, radius)
x, y = pos.get(node.id, (0, 0))
item.setPos(x, y)
self.scene.addItem(item)
items[node.id] = item
self._node_items.append(item)
sx += x
sy += y
for edge in graph.edges:
a, b = items.get(edge.source), items.get(edge.target)
if a and b:
e = _Edge(a, b, getattr(edge, "type", ""))
self.scene.addItem(e)
self._edge_items.append(e)
n = max(1, len(self._node_items))
self._centroid = QPointF(sx / n, sy / n)
self._fit()
if self.web is not None:
self._render_d3()
note = tr("structure.truncated_note") if getattr(graph, "truncated", False) else ""
self.status_message.emit(tr(
"structure.graph_summary", nodes=len(graph.nodes), edges=len(graph.edges), note=note))
self._preserve_answer()
def _render_d3(self) -> None:
if self.web is None or self._graph is None:
return
from ...core.d3_graph import build_html
try:
self.web.setHtml(build_html(self._graph), QUrl("https://cowork.local/"))
except Exception as exc:
self.status_message.emit(f"D3 view error: {exc}")
def _on_selection(self) -> None:
for item in self.scene.selectedItems():
if isinstance(item, _Node):
d = item.data
self.detail.setPlainText(f"[{d.kind.upper()}] {d.label}\n\n{d.detail}")
self._detail_mode = "node"
return
def _fit(self) -> None:
if self.web is not None and self._stack.currentWidget() is self.web:
self.web.page().runJavaScript("window.fitGraph && window.fitGraph();")
return
rect = self.scene.itemsBoundingRect()
if not rect.isNull():
self.view.fitInView(rect.adjusted(-40, -40, 40, 40), Qt.KeepAspectRatio)
def _export(self) -> None:
path, _ = QFileDialog.getSaveFileName(
self, tr("structure.export_title"), "structure-graph.png", "PNG (*.png)")
if not path:
return
showing_d3 = (self.web is not None and self._stack.currentWidget() is self.web)
if showing_d3:
self._export_d3_png(path)
else:
self._export_widget_grab(path)
def _export_d3_png(self, path: str) -> None:
def on_result(data_url) -> None:
if not isinstance(data_url, str) or "," not in data_url:
self._export_widget_grab(path)
return
import base64
try:
with open(path, "wb") as f:
f.write(base64.b64decode(data_url.split(",", 1)[1]))
self.status_message.emit(tr("structure.export_done", path=path))
except (OSError, ValueError) as exc:
self.status_message.emit(tr("structure.export_failed", err=str(exc)))
self.web.page().runJavaScript("window.exportPng ? window.exportPng() : ''", on_result)
def _export_widget_grab(self, path: str) -> None:
ok = self._stack.currentWidget().grab().save(path, "PNG")
if ok:
self.status_message.emit(tr("structure.export_done", path=path))
else:
self.status_message.emit(tr("structure.export_failed", err="grab() returned no image"))
@staticmethod
def _graph_context(graph) -> str:
from collections import defaultdict
by_kind = defaultdict(list)
for n in graph.nodes:
by_kind[n.kind].append(n.label)
lines = []
for kind in ("file", "class", "function", "method", "module", "section"):
items = by_kind.get(kind, [])
if items:
lines.append(f"{kind} ({len(items)}): " + ", ".join(items[:60]))
id2label = {n.id: n.label for n in graph.nodes}
rels = [f"{id2label.get(e.source, e.source)} -{e.type}-> {id2label.get(e.target, e.target)}"
for e in graph.edges[:140]]
if rels:
lines.append("Relationships (sample):\n" + "\n".join(rels))
return "\n".join(lines)[:7000]
presentation/graph/graph_scene.py
-138
View File
@@ -1,138 +0,0 @@
"""Các phần tử vẽ của đồ thị: node, cạnh, khung nhìn — R08-T14.
Thuần đồ hoạ Qt, không biết gì về GraphRAG hay agent. Tách riêng vì đây là
chỗ duy nhất cần mở khi chỉnh cách đồ thị trông ra sao — màu, hình mũi tên,
cách kéo thả và phóng to.
"""
from __future__ import annotations
import re
from pathlib import Path
from PySide6.QtCore import QObject, QPointF, Qt, Slot
from PySide6.QtGui import QBrush, QColor, QFont, QPen
from PySide6.QtWidgets import QGraphicsEllipseItem, QGraphicsLineItem, QGraphicsSimpleTextItem, QGraphicsView
from ...core.structure_graph import EDGE_KIND_COLORS, NODE_KIND_COLORS
from ...theme import current_palette
from ...i18n import tr
from ...ui.osutil import open_folder, open_location
class _Bridge(QObject):
"""Exposed to the D3 page so a Shift+click on a node can open its
storage folder/link (local path or URL — see osutil.open_location)."""
@Slot(str)
def openPath(self, path: str) -> None: # noqa: N802 - JS-facing name
if path:
open_location(path)
class _Edge(QGraphicsLineItem):
def __init__(self, a: "_Node", b: "_Node", type_: str = ""):
super().__init__()
self.a, self.b = a, b
self.type = type_
# Colour the edge by its RELATIONSHIP type (contains/defines/method/…),
# so the graph shows what each connection MEANS — falling back to the
# source node's tint for any untyped edge.
color = QColor(EDGE_KIND_COLORS.get(type_, "")) if type_ else QColor()
if not color.isValid():
color = a.brush().color().lighter(130)
self._color = color
self.setPen(QPen(color, 1.4))
self.setZValue(-1)
# A small label naming the relationship, shown at the edge midpoint.
self._label = None
if type_:
self._label = QGraphicsSimpleTextItem(type_, self)
self._label.setBrush(QBrush(color.lighter(140)))
f = QFont()
f.setPointSize(7)
self._label.setFont(f)
self._label.setZValue(0)
a.edges.append(self)
b.edges.append(self)
self.adjust()
def adjust(self) -> None:
pa, pb = self.a.scenePos(), self.b.scenePos()
self.setLine(pa.x(), pa.y(), pb.x(), pb.y())
if self._label is not None:
br = self._label.boundingRect()
self._label.setPos((pa.x() + pb.x()) / 2 - br.width() / 2,
(pa.y() + pb.y()) / 2 - br.height() / 2)
class _Node(QGraphicsEllipseItem):
def __init__(self, data, radius: int):
super().__init__(-radius, -radius, 2 * radius, 2 * radius)
self.data = data
self.edges = []
tok = current_palette()
# NODE_KIND_COLORS is a categorical data encoding (one hue per node
# kind), not UI chrome — it stays fixed across themes on purpose so a
# given kind is always the same colour. Only the chrome follows tokens.
color = QColor(NODE_KIND_COLORS.get(data.kind, tok.text_muted))
self.setBrush(QBrush(color))
self.setPen(QPen(color.darker(160), 1.5))
self.setFlags(
QGraphicsEllipseItem.ItemIsMovable
| QGraphicsEllipseItem.ItemIsSelectable
| QGraphicsEllipseItem.ItemSendsGeometryChanges
)
self.setZValue(1)
label = QGraphicsSimpleTextItem(data.label, self)
label.setBrush(QBrush(QColor(tok.text)))
label.setPos(radius + 3, -8)
def itemChange(self, change, value): # noqa: N802
if change == QGraphicsEllipseItem.ItemPositionHasChanged:
for edge in self.edges:
edge.adjust()
return super().itemChange(change, value)
class _GraphView(QGraphicsView):
def __init__(self, scene):
super().__init__(scene)
self.setDragMode(QGraphicsView.NoDrag)
self._panning = False
self._pan_start = QPointF()
def wheelEvent(self, e): # noqa: N802
self.scale(1.15 if e.angleDelta().y() > 0 else 1 / 1.15,
1.15 if e.angleDelta().y() > 0 else 1 / 1.15)
def mousePressEvent(self, e): # noqa: N802
if e.button() == Qt.LeftButton and self.itemAt(e.pos()) is None:
self._panning = True
self._pan_start = e.position()
self.setCursor(Qt.ClosedHandCursor)
e.accept()
return
super().mousePressEvent(e)
def mouseMoveEvent(self, e): # noqa: N802
if self._panning:
delta = e.position() - self._pan_start
self._pan_start = e.position()
self.horizontalScrollBar().setValue(int(self.horizontalScrollBar().value() - delta.x()))
self.verticalScrollBar().setValue(int(self.verticalScrollBar().value() - delta.y()))
e.accept()
return
super().mouseMoveEvent(e)
def mouseReleaseEvent(self, e): # noqa: N802
if self._panning:
self._panning = False
self.setCursor(Qt.ArrowCursor)
e.accept()
return
super().mouseReleaseEvent(e)
def mouseDoubleClickEvent(self, e): # noqa: N802
"""Double-click or Ctrl+click on a node opens its storage folder."""
item = self.itemAt(e.pos())
if isinstance(item, _Node) and getattr(item.data, "path", ""):
open_folder(item.data.path)
e.accept()
return
super().mouseDoubleClickEvent(e)
presentation/graph/graph_web.py
-38
View File
@@ -1,38 +0,0 @@
"""Có dùng được QtWebEngine hay không — R08-T14.
Cờ khả năng, tách riêng vì cả ``structure_graph_view.py`` lẫn
``graph_render.py`` đều phải hỏi. Để ở một trong hai thì file kia import
ngược lại — vòng import.
WebEngine là add-on tuỳ chọn của PySide6, và bản đóng gói one-file của
PyInstaller không chạy được nó; những trường hợp đó rơi về khung nhìn Qt.
"""
from __future__ import annotations
import sys
from pathlib import Path
def _frozen_onefile() -> bool:
"""True only for a PyInstaller ONEFILE build. Onefile extracts itself to a
temp dir (sys._MEIPASS under %TEMP%), where the QtWebEngine helper process
can't run — creating a QWebEngineView hard-crashes the app (reported as
"click the graph tab → app closes"). A ONEDIR build keeps _MEIPASS as the
``_internal`` folder right next to the exe, where WebEngine works fine, so
it keeps the full embedded D3 view."""
if not getattr(sys, "frozen", False):
return False
meipass = getattr(sys, "_MEIPASS", "")
if not meipass:
return False
try:
return Path(meipass).resolve().parent != Path(sys.executable).resolve().parent
except OSError: # can't tell → play safe: use the native fallback
return True
try: # WebEngine + WebChannel are optional PySide6 add-ons
from PySide6.QtWebEngineWidgets import QWebEngineView
from PySide6.QtWebChannel import QWebChannel
_HAS_WEB = not _frozen_onefile()
except Exception: # pragma: no cover
_HAS_WEB = False
presentation/scheduling/task_actions.py
-194
View File
@@ -1,194 +0,0 @@
"""Các thao tác trên một task: thêm, sửa, chạy ngay, xoá, xem log — R08-T11.
Tách khỏi ``ScheduleTaskTab`` để phần dựng bảng và phần hành động không nằm
lẫn nhau. ``_context_menu`` là chỗ tập trung: nó quyết định mục nào hiện ra
tuỳ theo đang chọn một hay nhiều thẻ.
Cùng kiểu mixin, xem ghi chú ở đầu ``presentation/shell/nav_rail.py``.
"""
from __future__ import annotations
# Import muộn trong hàm ở chỗ dùng: ba lớp này nằm cùng gói và một trong số
# chúng trộn ngược mixin này vào, nên import ở mức module là vòng.
import copy
from pathlib import Path
from typing import Dict, List, Optional
from PySide6.QtCore import Qt, Signal
from PySide6.QtWidgets import (
QAbstractItemView, QComboBox, QDialog, QDialogButtonBox, QHBoxLayout,
QLabel, QLineEdit, QListWidget, QListWidgetItem, QMenu, QMessageBox,
QPlainTextEdit, QPushButton, QScrollArea, QSizePolicy, QStackedWidget,
QTabBar, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget,
)
from ...core import tasks as taskrepo
from ...core.projects import list_projects
from ...core.tasks import STATUSES, chain_error, duplicate_task, new_task
from ...core.worker import AgentWorker
from ...i18n import on_language_changed, tr
from ...state import AppContext
from ...theme import current_palette
from ...ui.calendar_view import CalendarView
from ...ui.icons import icon
from ...ui.osutil import open_path
class TaskActionsMixin:
"""Thao tác trên task. Trộn vào ScheduleTaskTab."""
def _add_task_on_date(self, date_str: str) -> None:
"""Create a task pre-filled with the clicked calendar date (default
09:00) — same editor Add Task opens, nothing is saved until confirmed."""
from ...ui.task_editor_dialog import TaskEditorDialog
t = new_task("", schedule={"enabled": True, "run_at": f"{date_str} 09:00"})
dlg = TaskEditorDialog(t, taskrepo.list_tasks(self._tasks_dir), self, ctx=self.ctx)
if dlg.exec() and dlg.edited_task:
self._save_and_refresh(dlg.edited_task)
self.status_message.emit(tr("schedtask.msg_created"))
def _save_and_refresh(self, task: dict) -> None:
taskrepo.save_task(task, self._tasks_dir)
self.refresh()
def _add_task(self) -> None:
from ...ui.task_editor_dialog import TaskEditorDialog
dlg = TaskEditorDialog(None, taskrepo.list_tasks(self._tasks_dir), self, ctx=self.ctx)
if dlg.exec() and dlg.edited_task:
self._save_and_refresh(dlg.edited_task)
self.status_message.emit(tr("schedtask.msg_created"))
def _edit_task(self, task_id: str) -> None:
from ...ui.task_editor_dialog import TaskEditorDialog
task = taskrepo.load_task(task_id, self._tasks_dir)
if not task:
return
dlg = TaskEditorDialog(task, taskrepo.list_tasks(self._tasks_dir), self, ctx=self.ctx)
if dlg.exec() and dlg.edited_task:
self._save_and_refresh(dlg.edited_task)
def _on_double_click(self, item: QListWidgetItem) -> None:
tid = item.data(Qt.UserRole)
if tid:
self._edit_task(tid)
def _is_multi_selection(item, selected) -> bool:
"""True when the right-clicked card is part of an existing multi-item
selection — pure boolean, kept separate from _context_menu so it's
testable without ever invoking Qt's (modal, event-loop-blocking) menu."""
return len(selected) > 1 and item in selected
def _context_menu(self, col: _KanbanColumn, pos) -> None:
item = col.itemAt(pos)
if item is None or not item.data(Qt.UserRole):
return
selected = [it for it in col.selectedItems() if it.data(Qt.UserRole)]
if self._is_multi_selection(item, selected):
self._bulk_delete_menu(col, pos, selected)
return
tid = item.data(Qt.UserRole)
task = taskrepo.load_task(tid, self._tasks_dir)
if not task:
return
menu = QMenu(col)
run_act = menu.addAction(tr("schedtask.menu_run"))
edit_act = menu.addAction(tr("schedtask.menu_edit"))
dup_act = menu.addAction(tr("schedtask.menu_duplicate"))
paused = task.get("status") == "paused"
pause_act = menu.addAction(tr("schedtask.menu_resume" if paused else "schedtask.menu_pause"))
logs_act = menu.addAction(tr("schedtask.menu_logs"))
hist_act = menu.addAction(tr("schedtask.menu_history"))
next_act = menu.addAction(tr("schedtask.menu_create_next"))
menu.addSeparator()
del_act = menu.addAction(tr("schedtask.menu_delete"))
chosen = menu.exec(col.viewport().mapToGlobal(pos))
if chosen == run_act:
self._run_now(task)
elif chosen == edit_act:
self._edit_task(tid)
elif chosen == dup_act:
self._save_and_refresh(duplicate_task(task))
elif chosen == pause_act:
task["status"] = "backlog" if paused else "paused"
self._save_and_refresh(task)
elif chosen == logs_act:
self._view_logs(task)
elif chosen == hist_act:
from .run_history_dialog import _RunHistoryDialog
_RunHistoryDialog(task, self).exec()
elif chosen == next_act:
self._create_next_from_output(task)
elif chosen == del_act:
if QMessageBox.question(self, tr("schedtask.menu_delete"),
tr("schedtask.delete_confirm", title=task.get("title", ""))
) == QMessageBox.Yes:
taskrepo.delete_task(tid, self._tasks_dir)
self.refresh()
def _bulk_delete_menu(self, col: _KanbanColumn, pos, selected) -> None:
"""Right-click on a multi-selection within one column (Shift/Ctrl-click
several cards first): one action deletes every selected task. The
popup itself is a thin wrapper — see _confirm_and_delete_selected for
the actual (independently testable) confirm+delete logic."""
menu = QMenu(col)
del_act = menu.addAction(tr("schedtask.menu_delete_selected", n=len(selected)))
chosen = menu.exec(col.viewport().mapToGlobal(pos))
if chosen == del_act:
self._confirm_and_delete_selected(selected)
def _confirm_and_delete_selected(self, selected) -> bool:
"""Confirm, then delete every task in ``selected``. Split out of
_bulk_delete_menu so tests can drive it directly without having to
fake a real (modal, event-loop-blocking) QMenu popup."""
if QMessageBox.question(
self, tr("schedtask.menu_delete"),
tr("schedtask.delete_multi_confirm", n=len(selected))) != QMessageBox.Yes:
return False
for item in selected:
tid = item.data(Qt.UserRole)
if tid:
taskrepo.delete_task(tid, self._tasks_dir)
self.refresh()
return True
def _run_now(self, task: dict) -> None:
if task.get("task_type") == "manual":
self.status_message.emit(tr("schedtask.msg_manual_norun"))
return
if self.scheduler is None:
self.status_message.emit(tr("schedtask.msg_no_scheduler"))
return
if self.scheduler.run_now(task["task_id"]):
self.status_message.emit(tr("schedtask.msg_running", title=task.get("title", "")))
self.refresh()
def _view_logs(self, task: dict) -> None:
run_id = task.get("logs", {}).get("last_run_id")
if not run_id:
QMessageBox.information(self, tr("schedtask.menu_logs"), tr("schedtask.no_runs_yet"))
return
folder = taskrepo.ARTIFACTS_DIR / task["task_id"] / run_id
if folder.exists():
open_path(str(folder))
else:
QMessageBox.information(self, tr("schedtask.menu_logs"), tr("schedtask.no_runs_yet"))
def _create_next_from_output(self, task: dict) -> None:
"""Scaffold a follow-up task pre-wired to consume this task's output."""
nxt = new_task(tr("schedtask.next_of", title=task.get("title", "")))
nxt["task_type"] = "cowork"
nxt["input"]["mode"] = "previous_task_output"
nxt["input"]["previous_task_id"] = task["task_id"]
nxt["dependency"]["previous_task_id"] = task["task_id"]
err = chain_error(taskrepo.list_tasks(self._tasks_dir) + [nxt],
task["task_id"], nxt["task_id"])
if err:
QMessageBox.warning(self, tr("schedtask.g_dependency"), err)
return
taskrepo.save_task(nxt, self._tasks_dir)
task["dependency"]["next_task_id"] = nxt["task_id"]
task["dependency"]["pass_output_to_next"] = True
if task["dependency"].get("run_next_mode", "none") == "none":
task["dependency"]["run_next_mode"] = "run_after_success"
taskrepo.save_task(task, self._tasks_dir)
self.refresh()
self._edit_task(nxt["task_id"])
def _ai_create(self) -> None:
from .ai_task_creator_dialog import _AiCreateDialog
dlg = _AiCreateDialog(self.ctx, self)
if dlg.exec() and dlg.created_tasks:
for t in dlg.created_tasks:
taskrepo.save_task(t, self._tasks_dir)
self.refresh()
self.status_message.emit(tr("schedtask.msg_ai_created", n=len(dlg.created_tasks)))