from io import BytesIO, StringIO
from html import escape
import csv
import fitz # PyMuPDF
from docx import Document
from reportlab.lib.pagesizes import A4
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer
from reportlab.lib.styles import getSampleStyleSheet
def uploaded_file_to_text(uploaded_file):
"""
Convert uploaded .txt, .docx, or text-based .pdf to plain text.
"""
filename = uploaded_file.name.lower()
data = uploaded_file.getvalue()
if filename.endswith(".txt"):
return data.decode("utf-8", errors="ignore"), "txt"
if filename.endswith(".docx"):
doc = Document(BytesIO(data))
return extract_docx_text(doc), "docx"
if filename.endswith(".pdf"):
return extract_pdf_text(data), "pdf"
raise ValueError("Unsupported file type. Please upload .txt, .docx, or .pdf.")
def extract_docx_text(doc):
parts = []
for paragraph in iter_docx_paragraphs(doc):
text = paragraph.text.strip()
if text:
parts.append(text)
return "\n".join(parts)
def extract_pdf_text(pdf_bytes):
parts = []
with fitz.open(stream=pdf_bytes, filetype="pdf") as doc:
for page in doc:
text = page.get_text("text", sort=True)
if text.strip():
parts.append(text)
return "\n\n".join(parts)
def iter_docx_paragraphs(doc):
"""
Yield paragraphs from body, tables, headers and footers.
This does not cover every exotic Word object, but it is a good MVP.
"""
for paragraph in doc.paragraphs:
yield paragraph
for table in doc.tables:
yield from iter_table_paragraphs(table)
for section in doc.sections:
header_footer_parts = [
section.header,
section.footer,
section.first_page_header,
section.first_page_footer,
section.even_page_header,
section.even_page_footer,
]
for part in header_footer_parts:
for paragraph in part.paragraphs:
yield paragraph
for table in part.tables:
yield from iter_table_paragraphs(table)
def iter_table_paragraphs(table):
for row in table.rows:
for cell in row.cells:
for paragraph in cell.paragraphs:
yield paragraph
for nested_table in cell.tables:
yield from iter_table_paragraphs(nested_table)
def build_placeholder_replacements(text, analyze_results):
"""
Build stable placeholders from Presidio results.
Example:
John Smith -> [PERSON_01]
Rotterdam -> [LOCATION_01]
"""
counters = {}
replacements = {}
report_rows = []
sorted_results = sorted(analyze_results, key=lambda r: (r.start, r.end))
for result in sorted_results:
original = text[result.start:result.end].strip()
if not original:
continue
if original in replacements:
continue
entity_type = result.entity_type
counters[entity_type] = counters.get(entity_type, 0) + 1
placeholder = f"[{entity_type}_{counters[entity_type]:02d}]"
replacements[original] = placeholder
report_rows.append(
{
"entity_type": entity_type,
"detected_text": original,
"placeholder": placeholder,
"score": round(float(result.score), 3),
}
)
return replacements, report_rows
def apply_replacements_to_text(text, replacements):
"""
Apply longest replacements first to avoid partial replacement problems.
"""
output = text
for original, placeholder in sorted(
replacements.items(), key=lambda item: len(item[0]), reverse=True
):
output = output.replace(original, placeholder)
return output
def anonymized_docx_from_original(uploaded_file, replacements):
"""
Open original .docx and replace detected text inside existing paragraphs/runs.
This preserves most normal Word formatting, but not all complex Word features.
"""
doc = Document(BytesIO(uploaded_file.getvalue()))
for paragraph in iter_docx_paragraphs(doc):
replace_in_paragraph_runs(paragraph, replacements)
output = BytesIO()
doc.save(output)
output.seek(0)
return output.getvalue()
def replace_in_paragraph_runs(paragraph, replacements):
for original, placeholder in sorted(
replacements.items(), key=lambda item: len(item[0]), reverse=True
):
while replace_once_in_runs(paragraph, original, placeholder):
pass
def replace_once_in_runs(paragraph, old_text, new_text):
"""
Replace one occurrence across Word runs.
Word splits text into runs for formatting. This method tries to replace
text without rebuilding the whole paragraph.
"""
if not paragraph.runs:
return False
full_text = "".join(run.text for run in paragraph.runs)
start = full_text.find(old_text)
if start == -1:
return False
end = start + len(old_text)
current_pos = 0
replacement_done = False
for run in paragraph.runs:
run_text = run.text
run_start = current_pos
run_end = current_pos + len(run_text)
current_pos = run_end
if run_end <= start or run_start >= end:
continue
local_start = max(start - run_start, 0)
local_end = min(end - run_start, len(run_text))
before = run_text[:local_start]
after = run_text[local_end:]
if not replacement_done:
if end <= run_end:
run.text = before + new_text + after
else:
run.text = before + new_text
replacement_done = True
else:
if end <= run_end:
run.text = after
else:
run.text = ""
return True
def docx_from_text(text):
doc = Document()
for line in text.splitlines():
doc.add_paragraph(line)
output = BytesIO()
doc.save(output)
output.seek(0)
return output.getvalue()
def pdf_from_text(text):
output = BytesIO()
pdf = SimpleDocTemplate(output, pagesize=A4)
styles = getSampleStyleSheet()
normal = styles["Normal"]
story = []
for block in text.splitlines():
safe_block = escape(block) if block.strip() else " "
story.append(Paragraph(safe_block, normal))
story.append(Spacer(1, 6))
pdf.build(story)
output.seek(0)
return output.getvalue()
def replacement_report_csv(report_rows):
output = StringIO()
writer = csv.DictWriter(
output,
fieldnames=["entity_type", "detected_text", "placeholder", "score"],
)
writer.writeheader()
for row in report_rows:
writer.writerow(row)
return output.getvalue().encode("utf-8")