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

merge: hoà cập nhật mới từ origin/feature/delta-team/epic-R04 (R08 Chat UI Hub + R10)

Browse Source 
Đồng nghiệp đã push thêm 10 commit lên nhánh trong lúc đang xử lý merge
trước đó (R08-T01..T06 chat_panel.py split, R08 folder/dashboard/graph/
scheduling hoàn thiện, R10 CI Quality Gates + Contributor Recipes + E2E
smoke test). Resolve conflict:

- application/monitoring/__init__.py, domain/tasks/__init__.py,
  infrastructure/persistence/json/__init__.py: chỉ khác docstring — giữ bản
  HEAD (đầy đủ ngữ cảnh EPIC hơn), hợp nhất __all__ khi cần
  (MonitoringQueryService).
- tests/fakes/__init__.py: hợp nhất __getattr__ để lazy-load cả
  FakeToolExecutor lẫn ToolInvocation (bản HEAD thiếu ToolInvocation), bỏ
  entry "FakeClock" bị lặp trong __all__.
- tests/integration/test_routing_surfaces.py (deleted by them): khôi phục
  lại bản đã sửa ở lần merge trước — verify lại: API routing
  (RoutingApplicationService.resolve/_apply_routing/_apply_co4e_routing)
  không đổi sau khi chat_panel.py chuyển sang presentation/chat/*, 9/9 test
  vẫn pass trên code đã merge.

Ghi chú (không sửa, ngoài phạm vi merge): tests/fakes/__init__.py trên nhánh
remote export "ToolInvocation" từ fake_tool_executor.py nhưng class này đã
bị xoá nhầm từ commit chung 10739f1 (breakdown folder tree epic R01) — hiện
là dead code, không ai import, nhưng sẽ raise ImportError nếu có test nào
sau này thử dùng.

Đã chạy pytest tests/: 793 passed (không phát sinh fail mới so với lần
merge trước — 8 fail còn lại đều do môi trường sandbox: thiếu package
keyring, và tên thư mục checkout "cowork-local" thay vì "cowork_local"
khiến vài test spawn-subprocess không import được package).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
duylh19
2026-08-28 11:57:14 +09:00
co-authored by Claude Sonnet 5
parent c7784defc3 95b3b27578
commit 4f0010a971
51 changed files with 6708 additions and 3852 deletions
Download Patch File Download Diff File Expand all files Collapse all files
presentation/folder/ai_edit_runner.py
+325
View File
@@ -0,0 +1,325 @@
"""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
@@ -0,0 +1,140 @@
"""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
@@ -0,0 +1,114 @@
"""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
@@ -0,0 +1,115 @@
"""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))