DOCXpress/main.py at main · paradox-solver/DOCXpress · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
from fastapi import FastAPI, HTTPException, UploadFile, File, Request
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from pydantic import BaseModel
from typing import Tuple, Optional
from typing import Dict, Any, Optional
from pathlib import Path

import os
import uuid
import platformdirs

from src.changes_persister import ChangesPersister
from src.docx_parser import DocxParser
from src.pdf_converter import convert_docx_to_pdf
from src.utils import cleanup_temp_files
from src.utils import logger


app = FastAPI(
    title="DOCXPress",
    description="Online DOCX Document Editor with PDF Export",
    version="1.0.0",
)

# Configuration directory

DATA_DIR = Path(platformdirs.user_data_dir("DOCXpress", "Paradoxsolver"))
DATA_DIR.mkdir(parents=True, exist_ok=True)
TEMP_DIR = DATA_DIR / "temp"
TEMP_DIR.mkdir(parents=True, exist_ok=True)
TEMPLATE_FILE = TEMP_DIR / "template.docx"

BASE_DIR = Path(__file__).parent
TEMPLATE_DIR = BASE_DIR / "templates"
STATIC_DIR = BASE_DIR / "static"

# Mount static files and templates
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
templates = Jinja2Templates(directory=TEMPLATE_DIR)

# Store session data（Please use database for production environment）
sessions = {}

changes_persister = ChangesPersister(DATA_DIR)


class DocumentUpdate(BaseModel):
    changes: Dict[str, Dict[str, str]]
    session_id: str


@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
    """front page - Show upload page"""
    template_exists = TEMPLATE_FILE.exists()

    return templates.TemplateResponse(
        "upload.html", {"request": request, "template_exists": template_exists}
    )


# Edit page route
@app.get("/editor", response_class=HTMLResponse)
async def edit_document(request: Request):
    """Main editing page - Merge persistent changes intocontent，Clearchanges"""
    if not TEMPLATE_FILE.exists():
        return templates.TemplateResponse("upload.html", {"request": request})

    try:
        # parse DOCX document
        content = DocxParser.extract_content(str(TEMPLATE_FILE))
        session_id = str(uuid.uuid4())

        # 1. Load all modifications from persistent storage
        all_changes = changes_persister.get_all_changes()
        logger.debug(
            f"📖 Loaded from persistent storage {len(all_changes)} modifications"
        )
        logger.debug(f"📋 Modify content: {all_changes}")

        # 2. Merge changes intocontentmiddle，generate newcontent
        merged_content = DocxParser.merge_changes_into_content(content, all_changes)
        # 3. Store session data - changesremain empty
        sessions[session_id] = {
            "content": merged_content,  # Use merged content
            "changes": {},  # Clear，Front end from scratch
            "template_path": str(TEMPLATE_FILE),
            "original_content": content,  # Save original content，Used for comparison when exporting
            "loaded_changes_count": len(all_changes),
        }

        logger.debug(f"📊 Session initialization completed:")
        logger.debug(
            f"  Number of original content paragraphs: {len(content.get('paragraphs', []))}"
        )
        logger.debug(
            f"  Number of paragraphs after merging: {len(merged_content.get('paragraphs', []))}"
        )
        logger.debug(f"  Number of modifications loaded: {len(all_changes)}")
        # 4. Render the editor page
        return templates.TemplateResponse(
            "editor.html",
            {
                "request": request,
                "content": merged_content,  # Pass merged content
                "session_id": session_id,
                "has_saved_changes": len(all_changes) > 0,
            },
        )

    except Exception as e:
        import traceback

        traceback.print_exc()
        error_html = f"""
        <!DOCTYPE html>
        <html>
        <head>
            <title>mistake - DOCXPress</title>
            <style>
                body {{ font-family: Arial, sans-serif; margin: 0; padding: 20px; background: #f5f7fa; }}
                .error-container {{ max-width: 600px; margin: 100px auto; text-align: center; background: white; padding: 40px; border-radius: 10px; box-shadow: 0 4px 20px rgba(0,0,0,0.1); }}
                h1 {{ color: #dc3545; }}
                .btn {{ display: inline-block; padding: 10px 20px; background: #4CAF50; color: white; text-decoration: none; border-radius: 5px; margin-top: 20px; }}
                .btn:hover {{ background: #45a049; }}
            </style>
        </head>
        <body>
            <div class="error-container">
                <h1>Document processing error</h1>
                <p style="color: #dc3545;">{str(e)}</p>
                <a href="/" class="btn">Return to home page</a>
            </div>
        </body>
        </html>
        """
        return HTMLResponse(error_html)


@app.post("/upload")
async def upload_template(file: UploadFile = File(...)):
    """uploadDOCXtemplate"""
    if not file.filename.lower().endswith(".docx"):
        raise HTTPException(400, "Only supports .docx File format")

    try:
        content = await file.read()
        TEMPLATE_FILE.write_bytes(content)
        changes_persister.clear_all_changes()
        # Redirect to the editing page after successful upload
        from fastapi.responses import RedirectResponse

        return RedirectResponse(url="/editor", status_code=303)

    except Exception as e:
        raise HTTPException(500, f"File upload failed: {str(e)}")


@app.post("/api/save")
async def save_changes(update: DocumentUpdate):
    """Save changes to memory and persistent storage"""
    if update.session_id not in sessions:
        raise HTTPException(404, "The session does not exist or has expired")

    # 1. Update changes in memory
    sessions[update.session_id]["changes"].update(update.changes)

    # 2. Incremental updates to persistent storage
    try:
        changes_persister.update_changes(update.changes)
        message = f"saved {len(update.changes)} modifications to persistent storage"
        return {
            "status": "success",
            "message": message,
            "total_changes": len(sessions[update.session_id]["changes"]),
            "persisted": True,
        }

    except Exception as e:
        logger.debug(f"❌ Error saving to persistent storage: {e}")
        return {
            "status": "warning",
            "message": f"saved {len(update.changes)} modified into memory，But persistent storage failed: {str(e)}",
            "total_changes": len(sessions[update.session_id]["changes"]),
            "persisted": False,
        }


@app.get("/api/reset/{session_id}")
async def reset_session(session_id: str):
    """Reset session modifications"""
    if session_id in sessions:
        sessions[session_id]["changes"] = {}
        return {"status": "success", "message": "Edits have been reset"}
    raise HTTPException(404, "Session does not exist")


@app.get("/api/debug/sessions")
async def debug_sessions():
    """Debug interface：View all conversations"""
    session_info = {}
    for sid, data in sessions.items():
        session_info[sid] = {
            "changes_count": len(data.get("changes", {})),
            "template": data.get("template_path", "unknown"),
        }

    return {"total_sessions": len(sessions), "sessions": session_info}


@app.get("/health")
async def health_check():
    """health check endpoint"""
    return {"status": "healthy", "service": "DOCXPress", "version": "1.0.0"}


@app.post("/api/cleanup/{session_id}")
async def cleanup_session(session_id: str, delete_template: bool = False):
    """Clean session data（Optional deletion of template files）"""
    try:
        # Clean session data
        if session_id in sessions:
            del sessions[session_id]

        # If you request to delete a template file
        if delete_template and TEMPLATE_FILE.exists():
            TEMPLATE_FILE.unlink()
            return {
                "status": "success",
                "message": "Session and template files cleaned",
            }

        return {"status": "success", "message": "Session cleared"}

    except Exception as e:
        return {"status": "error", "message": f"Cleanup failed: {str(e)}"}


@app.get("/api/check-template")
async def check_template():
    """Check if template file exists"""
    return {
        "exists": TEMPLATE_FILE.exists(),
        "filename": TEMPLATE_FILE.name if TEMPLATE_FILE.exists() else None,
        "size": TEMPLATE_FILE.stat().st_size if TEMPLATE_FILE.exists() else 0,
    }


@app.get("/api/export/{session_id}")
async def export_document(session_id: str, format: str = "docx"):

    def apply_changes_and_generate_temp_file(
        pdf_conversion: bool = False,
    ) -> Tuple[Optional[str], Optional[str]]:
        """
        Apply changes and generate temporary files
        """
        try:
            # Apply changes and save to temporaryDOCX
            docx_temp_path = DocxParser.export_changes(
                str(TEMPLATE_FILE), changes_persister.get_all_changes()
            )

            logger.debug(f"✅ temporaryDOCXFile creation completed: {docx_temp_path}")

            # 2. in the case ofPDFConvert，createPDF
            if pdf_conversion:

                logger.debug("🔄 startPDFConvert...")
                pdf_temp_path = convert_docx_to_pdf(docx_temp_path)
                if (
                    not pdf_temp_path
                    or not os.path.exists(pdf_temp_path)
                    or os.path.getsize(pdf_temp_path) == 0
                ):
                    raise Exception("PDFConversion failed")

                logger.debug(
                    f"✅ PDFConversion successful！ temporaryPDFdocument: {pdf_temp_path}"
                )

                if docx_temp_path and os.path.exists(docx_temp_path):
                    cleanup_temp_files(docx_temp_path)

                # returnPDFpaths and sourcesDOCXpath（for cleaning）
                return pdf_temp_path, None
            else:
                # Return directlyDOCXpath
                return docx_temp_path, None

        except Exception as e:
            import traceback

            traceback.print_exc()
            logger.debug(f"❌ File generation failed: {e}")
            return None, f"File generation failed: {str(e)}"

    """Export document（supportdocx/pdf）"""
    # Verify format
    if format not in ["docx", "pdf"]:
        raise HTTPException(400, "Invalid format. Use 'docx' or 'pdf'")

    # Generate files
    pdf_conversion = format == "pdf"

    session_data = sessions[session_id]
    changes = session_data["changes"].copy()

    # 2. Automatically save current before exportingchangesto persistent storage
    if changes:
        logger.debug(
            "📝 Synchronize current modifications to persistent storage before exporting..."
        )
        changes_persister.update_changes(changes)

    temp_path, error_msg = apply_changes_and_generate_temp_file(
        pdf_conversion=pdf_conversion
    )

    if error_msg or not temp_path or not os.path.exists(temp_path):
        # Clean temporary files
        if temp_path and os.path.exists(temp_path):
            cleanup_temp_files(temp_path)
        raise HTTPException(500, "File generation failed")

    # Set file type and file name
    if format == "pdf":
        media_type = "application/pdf"
        filename = "document.pdf"
    else:
        media_type = (
            "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
        )
        filename = "edited_document.docx"

    return FileResponse(temp_path, media_type=media_type, filename=filename)


if __name__ == "__main__":
    import uvicorn

    # Make sure the directory exists
    for dir_path in [TEMPLATE_DIR, STATIC_DIR, STATIC_DIR / "css", STATIC_DIR / "js"]:
        dir_path.mkdir(parents=True, exist_ok=True)

    # Make sure the template file exists
    if not (TEMPLATE_DIR / "base.html").exists():
        logger.debug(
            "⚠️  warn: Template file does not exist，please make sure templates/ Directory containing all template files"
        )

    uvicorn.run(app, host="0.0.0.0", port=8000)