# latex_commands.py # -*- coding: utf-8 -*- from __future__ import annotations import io from typing import Optional, List import logging import discord from discord import app_commands from discord.ext import commands from constants import ( LATEX_TEX_ENGINE, LATEX_MAX_COMPILE_SECONDS, LATEX_MAX_ATTACH_MB, LATEX_MAX_ZIP_MB, OPENAI_MODEL, ALLOWED_ROLES, GUILD_ID ) from latex_functions import ( is_tex_attachment, compile_single_tex_bytes, compile_zip_to_zip, ask_openai_diagnosis ) class LatexModule(commands.Cog): def __init__(self, bot: commands.Bot, logger_name: str): self.bot = bot self.logger = logging.getLogger(logger_name) self.logger_name = logger_name # -------- /tex (PDF + diagnoza; log -> debug logger) -------- @commands.hybrid_command( nsfw=False, name="tex", description="Kompiluje załączone .tex; PDF-y odsyła. Błędy: diagnoza z OpenAI.", guild=None if GUILD_ID is None else discord.Object(id=GUILD_ID), ) @commands.has_any_role(*ALLOWED_ROLES) async def tex(self, ctx: commands.Context, include_source_excerpt: Optional[bool] = True): self.logger.info("LaTeX command invoked by %s", ctx.author) ch = ctx.message.channel async with ch.typing(): atts = [a for a in (ctx.message.attachments or []) if is_tex_attachment(a, LATEX_MAX_ATTACH_MB)] if not atts: return await ctx.reply(f"Dołącz .tex (≤{LATEX_MAX_ATTACH_MB} MiB).", mention_author=False) self.logger.debug("LaTeX attachments: %s", ", ".join(a.filename for a in atts)) files: List[discord.File] = [] parts: List[str] = [] for att in atts: try: data = await att.read() self.logger.info("LaTeX read %s bytes from %s", len(data), att.filename) res = await compile_single_tex_bytes(data, att.filename, LATEX_TEX_ENGINE, LATEX_MAX_COMPILE_SECONDS, logger=self.logger_name) self.logger.info("LaTeX result for %s: %s", att.filename, res) self.logger.info("LaTeX log (%s)\n%s", att.filename, (res["log_text"] or "")[-2000:]) if res["ok"] and res["pdf_bytes"]: b = io.BytesIO(res["pdf_bytes"]) b.seek(0) b.name = res["pdf_name"] files.append(discord.File(b, filename=res["pdf_name"])) parts.append(f"✅ `{att.filename}` → `{res['pdf_name']}`") else: self.logger.info("LaTeX compile failed for %s", att.filename) excerpt = "" if include_source_excerpt: try: excerpt = data.decode("utf-8", errors="ignore")[:4000] except Exception: excerpt = "" advice = await ask_openai_diagnosis(res["log_text"] or "", excerpt, OPENAI_MODEL, logger=self.logger_name) parts.append(f"❌ `{att.filename}` — błąd kompilacji.\nDiagnoza:\n{advice}") raise except Exception as e: self.logger.info("Exception %s: %s", att.filename, e) parts.append(f"⚠️ `{att.filename}` — wyjątek (szczegóły w logu).") raise content = f"**LaTeX** — {len(atts)} plik(ów). Model: `{OPENAI_MODEL}`\n\n" + "\n\n".join(parts) await ctx.reply(content if len(content)<1900 else content[:1900]+"…", files=files, mention_author=False) # -------- /latexclean (PDF only; log -> debug logger) -------- @commands.hybrid_command( nsfw=False, name="latexclean", description="Kompiluje .tex i odsyła TYLKO PDF (log w debug).", guild=None if GUILD_ID is None else discord.Object(id=GUILD_ID), ) @commands.has_any_role(*ALLOWED_ROLES) async def latexclean(self, ctx: commands.Context): ch = ctx.message.channel async with ch.typing(): atts = [a for a in (ctx.message.attachments or []) if is_tex_attachment(a, LATEX_MAX_ATTACH_MB)] if not atts: return await ctx.reply(f"Dołącz .tex (≤{LATEX_MAX_ATTACH_MB} MiB).", mention_author=False) files: List[discord.File] = [] parts: List[str] = [] for att in atts: try: data = await att.read() res = await compile_single_tex_bytes(data, att.filename, LATEX_TEX_ENGINE, LATEX_MAX_COMPILE_SECONDS, logger=self.logger_name) self.logger.debug("LaTeX CLEAN (%s)\n%s", att.filename, (res["log_text"] or "")[-2000:]) if res["ok"] and res["pdf_bytes"]: b = io.BytesIO(res["pdf_bytes"]) b.seek(0) b.name = res["pdf_name"] files.append(discord.File(b, filename=res["pdf_name"])) parts.append(f"✅ `{att.filename}`") else: parts.append(f"❌ `{att.filename}` — błąd (log w debug).") except Exception as e: self.logger.debug("CLEAN exception %s: %s", att.filename, e) parts.append(f"⚠️ `{att.filename}` — wyjątek (log w debug).") raise await ctx.reply("**LaTeX clean** — " + ", ".join(parts), files=files, mention_author=False) # -------- /texbatch (ZIP -> ZIP) -------- @app_commands.command( name="texbatch", description="Wyślij ZIP z .tex → odeślę ZIP z PDF-ami (tylko udane)." ) @app_commands.describe(archive=f"Archiwum .zip (max ~{LATEX_MAX_ZIP_MB} MiB)") @app_commands.checks.has_any_role(*ALLOWED_ROLES) async def texbatch(self, interaction: discord.Interaction, archive: discord.Attachment): await interaction.response.defer() if not archive or not archive.filename.lower().endswith(".zip"): return await interaction.followup.send("Dołącz ZIP (.zip) z plikami .tex.", ephemeral=True) if archive.size > LATEX_MAX_ZIP_MB * 1024 * 1024: return await interaction.followup.send(f"ZIP > {LATEX_MAX_ZIP_MB} MiB — za duży.", ephemeral=True) data = await archive.read() out_zip_bytes, failed = await compile_zip_to_zip(data, LATEX_TEX_ENGINE, LATEX_MAX_COMPILE_SECONDS, logger=self.logger_name) if not out_zip_bytes and failed: return await interaction.followup.send("Brak PDF-ów. " + "; ".join(failed[:10]), ephemeral=True) bio = io.BytesIO(out_zip_bytes) bio.seek(0) bio.name = "compiled_pdfs.zip" if failed: self.logger.debug("BATCH failed: %s", ", ".join(failed)) await interaction.followup.send("✅ Gotowe (ZIP w załączniku)." + (f" ❌ Błędy: {len(failed)}" if failed else ""), files=[discord.File(bio, filename=bio.name)]) async def setup(bot): logger = logging.getLogger("discord") await bot.add_cog(LatexModule(bot, logger_name="discord")) logger.info("Loading kinky latex module done")