import base64
import hashlib
import hmac
import secrets
import sqlite3
import os
import re
import statistics
import tempfile
import threading
import time
from pathlib import Path
import gradio as gr
import spaces
import torch
from docx import Document
from transformers import AutoModelForSeq2SeqLM, AutoTokenizer
# ---------------------------------------------------------------------------
# Brand (colours sampled from the CS Software Solutions logo)
# ---------------------------------------------------------------------------
BRAND = {
"navy": "#07228B",
"purple": "#47148B",
"magenta": "#E92E7F",
"red": "#E12B41",
"orange": "#FE5828",
"yellow": "#F5C518",
"green": "#50BF49",
"sky": "#50B4EF",
"grey": "#4A4A4A",
}
RAINBOW = (
f"linear-gradient(90deg, {BRAND['purple']}, {BRAND['navy']}, {BRAND['sky']}, "
f"{BRAND['green']}, {BRAND['yellow']}, {BRAND['orange']}, {BRAND['red']}, {BRAND['magenta']})"
)
APP_NAME = "CS Academic Paraphraser"
SITE_URL = "https://cssoftwaresolutions.in"
LOGO_PATH = Path(__file__).parent / "logo.png"
LOGO_SRC = ""
FAVICON_PATH = Path(__file__).parent / "favicon.png"
FAVICON_SRC = ""
if FAVICON_PATH.exists():
FAVICON_SRC = "data:image/png;base64," + base64.b64encode(FAVICON_PATH.read_bytes()).decode()
if LOGO_PATH.exists():
LOGO_SRC = "data:image/png;base64," + base64.b64encode(LOGO_PATH.read_bytes()).decode()
# ---------------------------------------------------------------------------
# Model (kept in CPU RAM while idle; moved to GPU only inside @spaces.GPU)
# ---------------------------------------------------------------------------
MODEL_ID = os.getenv("PARROT_MODEL", "prithivida/parrot_paraphraser_on_T5")
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
model = AutoModelForSeq2SeqLM.from_pretrained(MODEL_ID).eval()
model_lock = threading.Lock()
GPU_CHUNK = 48 # sentences per ZeroGPU allocation (short reservations)
CPU_CHUNK = 16 # sentences per CPU step (keeps progress updating)
MODES = ["Auto (GPU, falls back to CPU)", "CPU only (no GPU quota)"]
MAX_PASTE_WORDS = 6000 # keeps a single paste request within quota
# ---------------------------------------------------------------------------
# Text helpers
# ---------------------------------------------------------------------------
def split_sentences(text):
parts = re.split(r"(?<=[.!?])\s+(?=[A-Z0-9\"'(\[])", text.strip())
return [p.strip() for p in parts if p.strip()]
def should_process(text, min_words):
text = (text or "").strip()
if len(text.split()) < int(min_words):
return False
if re.fullmatch(r"https?://\S+|www\.\S+|\S+@\S+\.\S+", text):
return False
alpha = sum(ch.isalpha() for ch in text)
if alpha < max(5, int(len(text) * 0.35)):
return False
return True
def words(text):
return re.findall(r"[A-Za-z0-9']+", (text or "").lower())
# ---------------------------------------------------------------------------
# Tentative indicators (local heuristics — NOT Turnitin / iThenticate / GPTZero)
# ---------------------------------------------------------------------------
def _ngrams(tokens, n):
return {tuple(tokens[i:i + n]) for i in range(len(tokens) - n + 1)}
def similarity_to_source(original, revised):
"""Share of the revised text's word n-grams that also occur in the original."""
a, b = words(original), words(revised)
if len(b) < 4 or len(a) < 4:
return 0.0
scores, weights = [], []
for n, w in ((2, 0.2), (3, 0.35), (4, 0.45)):
gb = _ngrams(b, n)
if not gb:
continue
scores.append(len(gb & _ngrams(a, n)) / len(gb))
weights.append(w)
if not weights:
return 0.0
return 100.0 * sum(s * w for s, w in zip(scores, weights)) / sum(weights)
AI_MARKERS = [
"delve", "delves", "delving", "furthermore", "moreover", "additionally",
"in conclusion", "it is important to note", "it is worth noting", "plays a crucial role",
"plays a pivotal role", "pivotal", "crucial", "comprehensive", "leverage", "leveraging",
"seamless", "seamlessly", "robust", "landscape", "underscore", "underscores",
"notably", "in today's", "fostering", "foster", "multifaceted", "intricate",
"realm", "tapestry", "navigate", "navigating", "holistic", "paramount",
"a testament to", "shed light", "sheds light", "ever-evolving", "cutting-edge",
"in the realm of", "overall,", "ultimately", "enhance", "enhancing", "streamline",
"valuable insights", "key role", "vital role", "significantly",
]
def ai_likelihood(text):
"""Stylometric estimate (0–100) of machine-like writing."""
toks = words(text)
if len(toks) < 30:
return None
sents = split_sentences(text) or [text]
lengths = [len(words(s)) for s in sents if words(s)]
# 1. Burstiness: humans vary sentence length more than LLMs.
if len(lengths) >= 3:
cv = statistics.pstdev(lengths) / max(1.0, statistics.mean(lengths))
burst = min(1.0, max(0.0, (0.60 - cv) / 0.45))
else:
burst = 0.5
# 2. Density of stock "LLM" vocabulary per 100 words.
low = " " + " ".join(toks) + " "
raw = (text or "").lower()
hits = sum(raw.count(m) if " " in m or "," in m else low.count(f" {m} ") for m in AI_MARKERS)
marker = min(1.0, (hits / len(toks) * 100) / 2.5)
# 3. Uniformity: mid-length sentences, repeated openers, low lexical variety.
mean_len = statistics.mean(lengths)
band = 1.0 if 16 <= mean_len <= 28 else 0.4
openers = [words(s)[0] for s in sents if words(s)]
repeat = 1 - len(set(openers)) / len(openers) if openers else 0
window = toks[:400]
ttr = len(set(window)) / len(window)
variety = min(1.0, max(0.0, (0.62 - ttr) / 0.25))
uniform = 0.45 * band + 0.3 * min(1.0, repeat * 2) + 0.25 * variety
score = 100 * (0.40 * burst + 0.35 * marker + 0.25 * uniform)
return float(max(3.0, min(97.0, score)))
def words_changed(original, revised):
a, b = words(original), words(revised)
if not b:
return 0.0
return 100.0 * (1 - len(set(b) & set(a)) / len(set(b)))
def _level(value, invert=False):
if value is None:
return "#9aa0a6", "n/a"
v = 100 - value if invert else value
if v < 30:
return BRAND["green"], "Low"
if v < 55:
return BRAND["orange"], "Moderate"
return BRAND["red"], "High"
def _gauge(title, value, caption, invert=False, fixed_color=None):
color, label = _level(value, invert)
if fixed_color:
color = fixed_color
label = "Rewrite depth"
pct = 0 if value is None else round(value)
shown = "—" if value is None else f"{pct}%"
return f"""
{title}
{label}
{caption}
"""
def score_panel(original, revised):
sim = similarity_to_source(original, revised)
ai_in = ai_likelihood(original)
ai_out = ai_likelihood(revised)
changed = words_changed(original, revised)
return f"""
{_gauge("Similarity to source", sim, "Word n-gram overlap with your original")}
{_gauge("AI indicator · input", ai_in, "Stylometric estimate of the original")}
{_gauge("AI indicator · output", ai_out, "Stylometric estimate after paraphrasing")}
{_gauge("Vocabulary changed", changed, "Share of new wording", fixed_color=BRAND["navy"])}
Tentative indicators only. Similarity is measured against your own original text,
not against internet or journal databases, and the AI indicator is a stylometric heuristic.
Results will differ from Turnitin, iThenticate or other detectors. Always review the
output for meaning, technical terms and citations before use.
"""
EMPTY_SCORES = """
Scores appear here after paraphrasing — similarity to source, AI indicators and rewrite depth.
"""
# ---------------------------------------------------------------------------
# DOCX structure helpers
# ---------------------------------------------------------------------------
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 collect_paragraphs(doc):
seen = set()
def unique(paragraphs):
for paragraph in paragraphs:
key = id(paragraph._p)
if key not in seen:
seen.add(key)
yield paragraph
yield from unique(doc.paragraphs)
for table in doc.tables:
yield from unique(iter_table_paragraphs(table))
for section in doc.sections:
for part in (
section.header, section.footer,
section.first_page_header, section.first_page_footer,
section.even_page_header, section.even_page_footer,
):
yield from unique(part.paragraphs)
for table in part.tables:
yield from unique(iter_table_paragraphs(table))
def _is_protected(run):
return bool(
run._r.xpath(".//w:drawing")
or run._r.xpath(".//w:object")
or run._r.xpath(".//w:fldChar")
)
def replace_text_preserving_paragraph(paragraph, new_text):
runs = list(paragraph.runs)
if not runs:
paragraph.add_run(new_text)
return
target = next((r for r in runs if not _is_protected(r)), None)
if target is None:
target = paragraph.add_run()
target.text = new_text
for run in runs:
if run is not target and not _is_protected(run):
run.text = ""
# ---------------------------------------------------------------------------
# Paraphrasing
# ---------------------------------------------------------------------------
def _generate(texts, temperature, device, batch_size):
temperature = max(0.3, min(float(temperature), 1.5))
outputs = []
with torch.inference_mode():
for start in range(0, len(texts), batch_size):
chunk = texts[start:start + batch_size]
encoded = tokenizer(
["paraphrase: " + t for t in chunk],
return_tensors="pt", padding=True, truncation=True, max_length=256,
)
encoded = {k: v.to(device) for k, v in encoded.items()}
generated = model.generate(
**encoded,
do_sample=True,
temperature=temperature,
top_p=0.92,
num_return_sequences=1,
max_new_tokens=180,
no_repeat_ngram_size=3,
)
decoded = tokenizer.batch_decode(generated.detach().cpu(), skip_special_tokens=True)
outputs.extend(c.strip() if c.strip() else s for s, c in zip(chunk, decoded))
del encoded, generated
return outputs
def _gpu_duration(texts, temperature):
# Reserve only what this chunk needs, so visitors' daily ZeroGPU quota lasts longer.
return int(min(90, 12 + 0.3 * len(texts or [])))
@spaces.GPU(duration=_gpu_duration)
def paraphrase_batch_gpu(texts, temperature):
if not texts:
return []
if not torch.cuda.is_available():
raise RuntimeError("GPU allocation did not expose CUDA.")
with model_lock:
model.to("cuda")
try:
return _generate(texts, temperature, "cuda", batch_size=16)
finally:
model.to("cpu")
torch.cuda.empty_cache()
def paraphrase_batch_cpu(texts, temperature):
if not texts:
return []
with model_lock:
model.to("cpu")
return _generate(texts, temperature, "cpu", batch_size=4)
def _is_gpu_unavailable(err):
msg = str(err).lower()
return any(k in msg for k in ("quota", "zerogpu", "gpu task aborted", "cuda", "no gpu", "queue is full"))
def paraphrase_paragraphs(texts, temperature, progress=None, lo=0.15, hi=0.85, mode="Auto"):
"""Returns (paragraphs, engine_note)."""
groups, flat = [], []
for text in texts:
sents = split_sentences(text) or [text]
start = len(flat)
flat.extend(sents)
groups.append((start, len(flat)))
use_gpu = not str(mode).lower().startswith("cpu")
fell_back = False
rewritten = []
total = max(1, len(flat))
step = GPU_CHUNK if use_gpu else CPU_CHUNK
i = 0
while i < len(flat):
chunk = flat[i:i + step]
engine = "GPU" if use_gpu else "CPU"
if progress is not None:
progress(lo + (hi - lo) * i / total,
desc=f"Paraphrasing sentences {i + 1}–{i + len(chunk)} of {total} ({engine})")
if use_gpu:
try:
rewritten.extend(paraphrase_batch_gpu(chunk, temperature))
except Exception as err:
if not _is_gpu_unavailable(err):
raise
use_gpu, fell_back, step = False, True, CPU_CHUNK
continue # retry the same chunk on CPU
else:
rewritten.extend(paraphrase_batch_cpu(chunk, temperature))
i += len(chunk)
if fell_back:
note = "⚠️ GPU quota was unavailable, so the remaining text was processed on CPU (slower, same quality)."
elif not str(mode).lower().startswith("cpu"):
note = "⚡ Processed on GPU."
else:
note = "🖥️ Processed on CPU."
return [" ".join(rewritten[s:e]) for s, e in groups], note
def _preview(pairs, limit=10):
blocks = []
for i, (before, after) in enumerate(pairs[:limit], 1):
blocks.append(
f"#### Change {i}\n**Original:** {before}\n\n**Paraphrased:** {after}"
)
return "\n\n---\n\n".join(blocks) if blocks else "No textual changes were generated."
# ---------------------------------------------------------------------------
# Handlers
# ---------------------------------------------------------------------------
def _docx_core(file_path, temperature, min_words, mode, progress):
if not file_path:
raise gr.Error("Please upload a .docx file.")
source_path = Path(file_path)
if source_path.suffix.lower() != ".docx":
raise gr.Error("Only .docx Word documents are supported.")
progress(0.05, desc="Opening Word document")
doc = Document(str(source_path))
targets, originals = [], []
for paragraph in collect_paragraphs(doc):
text = paragraph.text
if should_process(text, min_words):
targets.append(paragraph)
originals.append(text)
if not originals:
raise gr.Error("No eligible prose paragraphs were found in the document.")
revised, engine_note = paraphrase_paragraphs(originals, temperature, progress, mode=mode)
progress(0.9, desc="Injecting revised text into the original layout")
changed, pairs, final_after = 0, [], []
for paragraph, before, after in zip(targets, originals, revised):
after = (after or "").strip()
if after and after != before.strip():
replace_text_preserving_paragraph(paragraph, after)
changed += 1
pairs.append((before, after))
final_after.append(after)
else:
final_after.append(before)
output_path = _save_output(doc, source_path.stem)
progress(1.0, desc="Completed")
summary = (
f"✅ **Completed** — {changed} of {len(originals)} eligible paragraphs rewritten. "
f"Layout, styles, tables, images and headers/footers were kept in place.\n\n{engine_note}"
)
scores = score_panel("\n".join(originals), "\n".join(final_after))
return output_path, download_button(output_path), summary, scores, _preview(pairs)
def _text_core(text, temperature, min_words, mode, progress):
text = (text or "").strip()
if not text:
raise gr.Error("Please paste some text to paraphrase.")
n_words = len(text.split())
if n_words > MAX_PASTE_WORDS:
raise gr.Error(
f"Pasted text has {n_words} words. Please keep it under {MAX_PASTE_WORDS} words, "
"or upload it as a .docx instead."
)
progress(0.05, desc="Reading pasted text")
paragraphs = [p.strip() for p in re.split(r"\n\s*\n|\r?\n", text)]
idx = [i for i, p in enumerate(paragraphs) if should_process(p, min_words)]
if not idx:
raise gr.Error(f"No paragraph has at least {int(min_words)} words of prose. Lower the minimum and try again.")
revised, engine_note = paraphrase_paragraphs([paragraphs[i] for i in idx], temperature, progress, mode=mode)
result = list(paragraphs)
pairs = []
for i, after in zip(idx, revised):
after = (after or "").strip()
if after and after != paragraphs[i]:
result[i] = after
pairs.append((paragraphs[i], after))
output_text = "\n\n".join(p for p in result if p)
progress(0.95, desc="Preparing download")
doc = Document()
for p in result:
if p:
doc.add_paragraph(p)
output_path = _save_output(doc, "pasted_text")
progress(1.0, desc="Completed")
summary = f"✅ **Completed** — {len(pairs)} of {len(idx)} eligible paragraphs rewritten ({n_words} words in).\n\n{engine_note}"
scores = score_panel(
"\n".join(paragraphs[i] for i in idx),
"\n".join(result[i] for i in idx),
)
return output_text, output_path, download_button(output_path), summary, scores
# ---------------------------------------------------------------------------
# Accounts (email + password, PBKDF2-hashed, SQLite)
# Stored in /data when persistent storage is enabled; otherwise next to app.py.
# ---------------------------------------------------------------------------
_DATA_DIR = Path("/data") if Path("/data").is_dir() and os.access("/data", os.W_OK) else Path(__file__).parent
USERS_DB = Path(os.getenv("USERS_DB", str(_DATA_DIR / "users.db")))
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
def _db():
con = sqlite3.connect(str(USERS_DB), timeout=10)
con.execute(
"CREATE TABLE IF NOT EXISTS users (email TEXT PRIMARY KEY, name TEXT, salt TEXT, "
"pw TEXT, created TEXT DEFAULT CURRENT_TIMESTAMP, last_login TEXT, runs INTEGER DEFAULT 0)"
)
cols = {r[1] for r in con.execute("PRAGMA table_info(users)")}
if "active" not in cols:
con.execute("ALTER TABLE users ADD COLUMN active INTEGER DEFAULT 1")
return con
# Admins: comma-separated emails in the ADMIN_EMAILS secret (Space settings → Variables and secrets).
ADMIN_EMAILS = {e.strip().lower() for e in os.getenv("ADMIN_EMAILS", "").split(",") if e.strip()}
def is_admin(user):
return bool(user and user.get("email", "").lower() in ADMIN_EMAILS)
def _hash(password, salt):
return hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), 200_000).hex()
def create_user(name, email, password, confirm):
name, email = (name or "").strip(), (email or "").strip().lower()
if not name:
return None, "Please enter your name."
if not _EMAIL_RE.match(email):
return None, "Please enter a valid email address."
if len(password or "") < 8:
return None, "Password must be at least 8 characters."
if password != confirm:
return None, "Passwords do not match."
salt = secrets.token_hex(16)
with _db() as con:
if con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone():
return None, "An account with this email already exists. Please log in."
con.execute("INSERT INTO users (email, name, salt, pw, last_login) VALUES (?,?,?,?,CURRENT_TIMESTAMP)",
(email, name, salt, _hash(password, salt)))
return {"email": email, "name": name}, None
def check_user(email, password):
email = (email or "").strip().lower()
with _db() as con:
row = con.execute("SELECT name, salt, pw, active FROM users WHERE email=?", (email,)).fetchone()
if not row or not hmac.compare_digest(row[2], _hash(password or "", row[1])):
return None, "Incorrect email or password."
if row[3] == 0:
return None, "This account has been disabled. Please contact CS Software Solutions."
con.execute("UPDATE users SET last_login=CURRENT_TIMESTAMP WHERE email=?", (email,))
return {"email": email, "name": row[0]}, None
def _count_run(user):
try:
with _db() as con:
con.execute("UPDATE users SET runs = runs + 1 WHERE email=?", (user["email"],))
except Exception:
pass
def _require(user):
if not user or not user.get("email"):
raise gr.Error("Please log in to use the paraphraser.")
with _db() as con:
row = con.execute("SELECT active FROM users WHERE email=?", (user["email"],)).fetchone()
if not row or row[0] == 0:
raise gr.Error("This account has been disabled.")
def _msg(text, ok=False):
cls = "cs-ok" if ok else "cs-bad"
return f'{text}
' if text else ""
def _welcome(user):
return (f'👋 Signed in as {user["name"]} '
f'({user["email"]})
')
def _signed_in(user):
admin = is_admin(user)
return (user, gr.update(visible=False), gr.update(visible=True), _welcome(user), "",
gr.update(visible=admin), admin_stats(user) if admin else "", admin_table(user) if admin else None)
def do_login(email, password):
user, err = check_user(email, password)
if err:
return None, gr.update(), gr.update(), "", _msg(err), gr.update(visible=False), "", None
return _signed_in(user)
def do_signup(name, email, password, confirm):
user, err = create_user(name, email, password, confirm)
if err:
return None, gr.update(), gr.update(), "", _msg(err), gr.update(visible=False), "", None
return _signed_in(user)
def do_logout():
return (None, gr.update(visible=True), gr.update(visible=False), "",
_msg("You have been logged out.", ok=True), gr.update(visible=False), "", None)
# ---------------------------------------------------------------------------
# Admin panel
# ---------------------------------------------------------------------------
ADMIN_COLS = ["Name", "Email", "Joined", "Last login", "Runs", "Status", "Role"]
def _need_admin(user):
if not is_admin(user):
raise gr.Error("Admin access only.")
def admin_stats(user):
_need_admin(user)
with _db() as con:
total = con.execute("SELECT COUNT(*) FROM users").fetchone()[0]
active = con.execute("SELECT COUNT(*) FROM users WHERE active=1").fetchone()[0]
runs = con.execute("SELECT COALESCE(SUM(runs),0) FROM users").fetchone()[0]
new7 = con.execute("SELECT COUNT(*) FROM users WHERE created >= datetime('now','-7 days')").fetchone()[0]
seen7 = con.execute("SELECT COUNT(*) FROM users WHERE last_login >= datetime('now','-7 days')").fetchone()[0]
tiles = [("Total users", total), ("Active accounts", active), ("Total runs", runs),
("New (7 days)", new7), ("Logged in (7 days)", seen7)]
cells = "".join(f'' for k, v in tiles)
return f'{cells}
'
def admin_table(user, query=""):
_need_admin(user)
q = f"%{(query or '').strip().lower()}%"
with _db() as con:
rows = con.execute(
"SELECT name, email, created, last_login, runs, active FROM users "
"WHERE lower(name) LIKE ? OR lower(email) LIKE ? ORDER BY created DESC", (q, q)).fetchall()
return [[n, e, (c or "")[:16], (l or "")[:16], r, "Active" if a else "Disabled",
"Admin" if e in ADMIN_EMAILS else "User"] for n, e, c, l, r, a in rows]
def _admin_refresh(user, query, note, ok=True):
return admin_stats(user), admin_table(user, query), _msg(note, ok=ok)
def admin_search(user, query):
_need_admin(user)
return admin_table(user, query)
def admin_set_active(user, email, query, active):
_need_admin(user)
email = (email or "").strip().lower()
if not email:
return _admin_refresh(user, query, "Enter the user's email first.", ok=False)
if email in ADMIN_EMAILS and not active:
return _admin_refresh(user, query, "Admin accounts can't be disabled here.", ok=False)
with _db() as con:
n = con.execute("UPDATE users SET active=? WHERE email=?", (1 if active else 0, email)).rowcount
if not n:
return _admin_refresh(user, query, f"No user found with {email}.", ok=False)
return _admin_refresh(user, query, f"{email} is now {'active' if active else 'disabled'}.")
def admin_reset_pw(user, email, new_pw, query):
_need_admin(user)
email = (email or "").strip().lower()
if len(new_pw or "") < 8:
return _admin_refresh(user, query, "New password must be at least 8 characters.", ok=False)
salt = secrets.token_hex(16)
with _db() as con:
n = con.execute("UPDATE users SET salt=?, pw=? WHERE email=?", (salt, _hash(new_pw, salt), email)).rowcount
if not n:
return _admin_refresh(user, query, f"No user found with {email}.", ok=False)
return _admin_refresh(user, query, f"Password reset for {email}. Share it with them securely.")
def admin_delete(user, email, confirm, query):
_need_admin(user)
email = (email or "").strip().lower()
if (confirm or "").strip().lower() != email or not email:
return _admin_refresh(user, query, "Type the same email in the confirm box to delete.", ok=False)
if email in ADMIN_EMAILS:
return _admin_refresh(user, query, "Admin accounts can't be deleted here.", ok=False)
with _db() as con:
n = con.execute("DELETE FROM users WHERE email=?", (email,)).rowcount
if not n:
return _admin_refresh(user, query, f"No user found with {email}.", ok=False)
return _admin_refresh(user, query, f"Deleted {email}.")
def admin_export(user):
_need_admin(user)
import csv
folder = Path(tempfile.mkdtemp(dir=OUT_DIR))
path = folder / "users.csv"
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(ADMIN_COLS)
w.writerows(admin_table(user))
return str(path)
# ---------------------------------------------------------------------------
# Single branded progress panel (replaces Gradio's per-component loaders)
# ---------------------------------------------------------------------------
class _Reporter:
def __init__(self):
self.frac, self.desc, self.start = 0.0, "Starting…", time.time()
def __call__(self, frac, desc=None):
self.frac = max(self.frac, min(1.0, float(frac)))
if desc:
self.desc = desc
def _md2html(md):
html = re.sub(r"\*\*(.+?)\*\*", r"\1", md)
return html.replace("\n\n", "
").replace("\n", "
")
def status_idle(msg):
return f'{_md2html(msg)}
'
def status_done(msg):
return f'{_md2html(msg)}
'
def status_error(msg):
return f'⚠️ {msg}
'
def status_running(rep):
pct = round(rep.frac * 100)
secs = int(time.time() - rep.start)
return f"""
{rep.desc}
{pct}% · {secs}s
"""
def _run_live(core, args, n_outputs, status_index):
"""Run `core` in a thread and stream only the status panel until it finishes."""
rep = _Reporter()
box = {}
def work():
try:
box["result"] = core(*args, rep)
except Exception as err: # re-raised in the request thread
box["error"] = err
t = threading.Thread(target=work, daemon=True)
t.start()
keep = [gr.update()] * n_outputs
while t.is_alive():
frame = list(keep)
frame[status_index] = status_running(rep)
yield tuple(frame)
t.join(timeout=0.6)
if "error" in box:
err = box["error"]
frame = list(keep)
frame[status_index] = status_error(str(err) or "Something went wrong.")
yield tuple(frame)
if isinstance(err, gr.Error):
raise err
raise gr.Error(f"Processing failed: {err}")
yield box["result"]
def process_docx(file_path, temperature, min_words, mode=MODES[0], user=None):
_require(user)
_count_run(user)
if not file_path:
raise gr.Error("Please upload a .docx file.")
for frame in _run_live(_docx_core, (file_path, temperature, min_words, mode), 5, 2):
if isinstance(frame[2], str) and frame[2].lstrip().startswith("✅"):
frame = (frame[0], frame[1], status_done(frame[2]), frame[3], frame[4])
yield frame
def process_text(text, temperature, min_words, mode=MODES[0], user=None):
_require(user)
_count_run(user)
if not (text or "").strip():
raise gr.Error("Please paste some text to paraphrase.")
for frame in _run_live(_text_core, (text, temperature, min_words, mode), 5, 3):
if isinstance(frame[3], str) and frame[3].lstrip().startswith("✅"):
frame = (frame[0], frame[1], frame[2], status_done(frame[3]), frame[4])
yield frame
OUT_DIR = Path(tempfile.gettempdir()) / "cs_outputs"
OUT_DIR.mkdir(parents=True, exist_ok=True)
def _save_output(doc, stem):
"""Save to a unique folder with a clean, readable file name."""
safe = re.sub(r"[^A-Za-z0-9_-]+", "_", stem).strip("_") or "document"
folder = Path(tempfile.mkdtemp(dir=OUT_DIR))
path = folder / f"{safe}_paraphrased.docx"
doc.save(str(path))
return str(path)
def download_button(path):
"""Direct download link (works even if the file widget is blocked)."""
if not path or not Path(path).exists():
return ""
data = base64.b64encode(Path(path).read_bytes()).decode()
mime = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
name = Path(path).name
return (f''
f'⬇️ Download {name}')
def count_words(text):
n = len((text or "").split())
color = BRAND["red"] if n > MAX_PASTE_WORDS else BRAND["grey"]
return f"{n:,} / {MAX_PASTE_WORDS:,} words
"
# ---------------------------------------------------------------------------
# UI
# ---------------------------------------------------------------------------
CSS = f"""
:root {{
--cs-navy:{BRAND['navy']}; --cs-purple:{BRAND['purple']}; --cs-magenta:{BRAND['magenta']};
--cs-sky:{BRAND['sky']}; --cs-rainbow:{RAINBOW};
}}
.gradio-container {{ max-width: 1180px !important; margin: 0 auto !important;
font-family: 'Poppins', ui-sans-serif, system-ui, sans-serif !important; }}
body, .gradio-container {{ background: #f6f7fb !important; }}
.cs-header {{ display:flex; align-items:center; gap:22px; padding:22px 26px; border-radius:20px;
background:#fff; box-shadow:0 8px 30px rgba(7,34,139,.08); position:relative; overflow:hidden; }}
.cs-header::after {{ content:""; position:absolute; left:0; right:0; bottom:0; height:5px; background:var(--cs-rainbow); }}
.cs-header img {{ width:92px; height:92px; border-radius:50%; flex:none;
box-shadow:0 4px 14px rgba(0,0,0,.12); }}
.cs-header h1 {{ margin:0; font-size:30px; font-weight:800; color:var(--cs-navy); letter-spacing:-.3px; line-height:1.15; }}
.cs-header h1 span {{ color:var(--cs-magenta) !important; }}
.cs-header p {{ margin:6px 0 0; color:#555; font-size:14.5px; }}
.cs-tag {{ margin-top:8px; font-size:11.5px; letter-spacing:3px; color:#444; font-weight:600; }}
.cs-tag b {{ color:var(--cs-magenta); }} .cs-tag i {{ color:var(--cs-navy); font-style:normal; }}
@media (max-width:640px) {{ .cs-header {{ flex-direction:column; text-align:center; }} .cs-header h1 {{ font-size:24px; }} }}
.cs-card {{ background:#fff !important; border-radius:18px !important; padding:18px !important;
box-shadow:0 6px 24px rgba(7,34,139,.06) !important; border:1px solid #eceef6 !important; }}
button.primary, .cs-go {{ background:linear-gradient(90deg,var(--cs-navy),var(--cs-purple) 60%,var(--cs-magenta)) !important;
color:#fff !important; border:none !important; font-weight:700 !important; font-size:16px !important;
border-radius:12px !important; box-shadow:0 6px 18px rgba(71,20,139,.28) !important; }}
button.primary:hover {{ filter:brightness(1.08); transform:translateY(-1px); }}
.tabs button[role="tab"] {{ font-weight:600 !important; font-size:15px !important; }}
.tabs button[role="tab"].selected {{ color:var(--cs-navy) !important; border-color:var(--cs-magenta) !important; }}
.cs-scores {{ display:grid; grid-template-columns:repeat(4,1fr); gap:14px; }}
@media (max-width:820px) {{ .cs-scores {{ grid-template-columns:repeat(2,1fr); }} }}
.cs-scores.cs-empty {{ display:block; text-align:center; color:#888; padding:26px; border:2px dashed #e3e6f2; border-radius:16px; }}
.cs-gauge {{ background:#fff; border:1px solid #eceef6; border-radius:16px; padding:16px 10px; text-align:center;
box-shadow:0 4px 16px rgba(7,34,139,.05); }}
.cs-ring {{ width:104px; height:104px; margin:0 auto 10px; border-radius:50%;
background:conic-gradient(var(--c) calc(var(--p)*1%), #edf0f7 0); display:grid; place-items:center; }}
.cs-ring-inner {{ width:80px; height:80px; border-radius:50%; background:#fff; display:grid; place-items:center; }}
.cs-ring-inner span {{ font-size:22px; font-weight:800; color:#222; }}
.cs-gauge-title {{ font-weight:700; color:var(--cs-navy); font-size:14px; }}
.cs-chip {{ display:inline-block; margin:6px 0; padding:2px 12px; border-radius:99px; font-size:12px; font-weight:700; }}
.cs-gauge-cap {{ font-size:11.5px; color:#777; line-height:1.35; }}
.cs-disclaimer {{ margin-top:12px; font-size:12px; color:#666; background:#fff8ec; border-left:4px solid {BRAND['orange']};
padding:10px 14px; border-radius:10px; }}
.cs-count {{ text-align:right; font-size:12px; margin-top:-6px; }}
.cs-footer {{ text-align:center; margin-top:18px; padding:16px; color:#666; font-size:13px; }}
.cs-footer .bar {{ height:3px; width:220px; margin:0 auto 12px; border-radius:3px; background:var(--cs-rainbow); }}
.cs-footer a {{ color:var(--cs-navy); font-weight:700; text-decoration:none; }}
footer {{ display:none !important; }}
/* Custom single progress panel; Gradio's own loaders are switched off. */
.cs-status {{ padding:16px 18px; border-radius:14px; font-size:14.5px; line-height:1.5; }}
.cs-idle {{ color:#666; }}
.cs-done {{ background:#f1fbf1; border-left:4px solid {BRAND['green']}; color:#234; }}
.cs-err {{ background:#fff3f3; border-left:4px solid {BRAND['red']}; color:#8a1c1c; }}
.cs-run {{ text-align:center; background:#fff; border:1px solid #eceef6; }}
.cs-spin-logo {{ width:64px; height:64px; margin:4px auto 12px;
background:url("{FAVICON_SRC}") center/contain no-repeat; animation:cs-spin 1.4s linear infinite; }}
.cs-st-text {{ font-weight:600; color:var(--cs-navy); margin-bottom:10px; }}
.cs-bar {{ height:10px; border-radius:99px; background:#edf0f7; overflow:hidden; }}
.cs-bar > div {{ height:100%; border-radius:99px; background:var(--cs-rainbow); transition:width .5s ease; }}
.cs-st-meta {{ margin-top:8px; font-size:12px; color:#777; }}
@keyframes cs-spin {{ to {{ transform:rotate(360deg); }} }}
#huggingface-space-header, .show-api {{ display:none !important; }}
.cs-dl {{ display:block; text-align:center; margin-top:10px; padding:12px 16px; border-radius:12px; font-weight:700;
color:#fff !important; text-decoration:none !important;
background:linear-gradient(90deg,var(--cs-navy),var(--cs-purple) 60%,var(--cs-magenta)); }}
.cs-dl:hover {{ filter:brightness(1.08); }}
.cs-settings {{ background:#f8f9fd !important; border:1px solid #eceef6 !important; border-radius:14px !important;
padding:12px 14px !important; margin:10px 0 !important; }}
.cs-settings-title {{ font-weight:700; color:var(--cs-navy); font-size:15px; margin:0 0 4px; }}
.cs-stats {{ display:grid; grid-template-columns:repeat(5,1fr); gap:12px; }}
@media (max-width:820px) {{ .cs-stats {{ grid-template-columns:repeat(2,1fr); }} }}
.cs-stat {{ background:#fff; border:1px solid #eceef6; border-radius:14px; padding:14px; text-align:center;
box-shadow:0 4px 16px rgba(7,34,139,.05); }}
.cs-stat .n {{ font-size:26px; font-weight:800; color:var(--cs-navy); }}
.cs-stat .l {{ font-size:12px; color:#777; margin-top:2px; }}
.cs-auth {{ max-width:460px; margin:26px auto 0 !important; }}
.cs-auth-title {{ text-align:center; font-weight:800; font-size:22px; color:var(--cs-navy); margin:4px 0 2px; }}
.cs-auth-sub {{ text-align:center; color:#666; font-size:13.5px; margin-bottom:10px; }}
.cs-authmsg {{ padding:10px 14px; border-radius:10px; font-size:13.5px; margin-top:6px; }}
.cs-ok {{ background:#f1fbf1; color:#1e5b1e; border-left:4px solid {BRAND['green']}; }}
.cs-bad {{ background:#fff3f3; color:#8a1c1c; border-left:4px solid {BRAND['red']}; }}
.cs-userbar {{ display:flex; align-items:center; padding:10px 4px; color:#444; font-size:14px; }}
.cs-userbar small {{ color:#888; }}
.cs-logout {{ max-width:130px !important; margin-left:auto !important; }}
"""
FORCE_LIGHT = """
() => {
const url = new URL(window.location);
if (url.searchParams.get('__theme') !== 'light') {
url.searchParams.set('__theme', 'light');
window.location.href = url.href;
return;
}
// Hide the platform's floating Space badge (it can be injected after load).
const hide = () => {
const h = document.getElementById('huggingface-space-header');
if (h) h.style.setProperty('display', 'none', 'important');
};
hide();
new MutationObserver(hide).observe(document.documentElement, { childList: true, subtree: true });
// Brand favicon (replaces the platform's default tab icon).
const ICON = "__FAVICON__";
if (ICON) {
document.querySelectorAll("link[rel*='icon']").forEach(l => l.remove());
const link = document.createElement('link');
link.rel = 'icon'; link.type = 'image/png'; link.href = ICON;
document.head.appendChild(link);
}
}
"""
FORCE_LIGHT = FORCE_LIGHT.replace("__FAVICON__", FAVICON_SRC)
logo_html = f'
' if LOGO_SRC else ""
HEADER = f"""
"""
FOOTER = f"""
"""
theme = gr.themes.Soft(
primary_hue="indigo",
secondary_hue="pink",
neutral_hue="slate",
font=[gr.themes.GoogleFont("Poppins"), "ui-sans-serif", "sans-serif"],
radius_size="lg",
)
with gr.Blocks(title=f"{APP_NAME} · CS Software Solutions", theme=theme, css=CSS, js=FORCE_LIGHT) as demo:
user_state = gr.State(None)
gr.HTML(HEADER)
# ======================= Login / Sign up =======================
with gr.Column(visible=True, elem_classes="cs-auth") as auth_view:
with gr.Column(elem_classes="cs-card"):
gr.HTML('Welcome
'
'Log in or create a free account to start paraphrasing.
')
with gr.Tabs():
with gr.Tab("🔐 Log in"):
li_email = gr.Textbox(label="Email", placeholder="you@university.edu")
li_pw = gr.Textbox(label="Password", type="password")
li_btn = gr.Button("Log in", variant="primary", size="lg")
with gr.Tab("📝 Sign up"):
su_name = gr.Textbox(label="Full name")
su_email = gr.Textbox(label="Email", placeholder="you@university.edu")
su_pw = gr.Textbox(label="Password", type="password", info="At least 8 characters")
su_pw2 = gr.Textbox(label="Confirm password", type="password")
su_btn = gr.Button("Create account", variant="primary", size="lg")
auth_msg = gr.HTML()
# ======================= Main app (after login) =======================
with gr.Column(visible=False) as app_view:
with gr.Row(equal_height=True):
user_bar = gr.HTML()
logout_btn = gr.Button("Log out", size="sm", elem_classes="cs-logout")
with gr.Tabs():
# ---------------- Upload DOCX ----------------
with gr.Tab("📄 Upload Word (.docx)"):
with gr.Row(equal_height=False):
with gr.Column(scale=5, elem_classes="cs-card"):
input_file = gr.File(label="Word document (.docx)", file_types=[".docx"], type="filepath")
with gr.Group(elem_classes="cs-settings"):
gr.HTML('⚙️ Settings
')
d_temp = gr.Slider(0.3, 1.5, value=0.9, step=0.1, label="Paraphrase variation",
info="Higher = bolder rewording, lower = closer to the original")
d_min = gr.Slider(5, 30, value=8, step=1, label="Minimum words per paragraph",
info="Shorter lines (headings, captions) are left untouched")
d_mode = gr.Radio(MODES, value=MODES[0], label="Processing",
info="GPU is fast but limited by a daily usage quota; CPU is slower but always available")
d_btn = gr.Button("✨ Paraphrase document", variant="primary", size="lg")
gr.Markdown(
"Keeps sections, margins, styles, tables, images, numbering and headers/footers. "
"Mixed bold/italic inside one paragraph takes the first run's formatting."
)
with gr.Column(scale=5, elem_classes="cs-card"):
d_status = gr.HTML(status_idle("Upload a document and press **Paraphrase document**."))
d_out = gr.File(label="Download paraphrased .docx", interactive=False)
d_dl = gr.HTML()
d_scores = gr.HTML(EMPTY_SCORES)
with gr.Accordion("Preview of changes (first 10)", open=False):
d_preview = gr.Markdown()
# ---------------- Paste text ----------------
with gr.Tab("✍️ Paste text"):
with gr.Row(equal_height=True):
with gr.Column(scale=1, elem_classes="cs-card"):
p_in = gr.Textbox(label="Your text", lines=16, max_lines=30,
placeholder="Paste your paragraphs here. Separate paragraphs with a blank line.")
p_count = gr.HTML(count_words(""))
with gr.Column(scale=1, elem_classes="cs-card"):
p_out = gr.Textbox(label="Paraphrased text", lines=16, max_lines=30,
show_copy_button=True, interactive=False)
with gr.Row():
with gr.Column(scale=5):
with gr.Group(elem_classes="cs-settings"):
gr.HTML('⚙️ Settings
')
p_temp = gr.Slider(0.3, 1.5, value=0.9, step=0.1, label="Paraphrase variation",
info="Higher = bolder rewording, lower = closer to the original")
p_min = gr.Slider(3, 30, value=6, step=1, label="Minimum words per paragraph")
p_mode = gr.Radio(MODES, value=MODES[0], label="Processing")
p_btn = gr.Button("✨ Paraphrase text", variant="primary", size="lg")
with gr.Column(scale=5):
p_status = gr.HTML(status_idle("Paste text and press **Paraphrase text**."))
p_file = gr.File(label="Download as .docx", interactive=False)
p_dl = gr.HTML()
p_scores = gr.HTML(EMPTY_SCORES)
# ---------------- Admin (admins only) ----------------
with gr.Tab("🛡️ Admin", visible=False) as admin_tab:
ad_stats = gr.HTML()
with gr.Row():
ad_query = gr.Textbox(label="Search users", placeholder="Name or email", scale=4)
ad_refresh = gr.Button("🔄 Refresh", scale=1)
ad_export = gr.Button("⬇️ Export CSV", scale=1)
ad_table = gr.Dataframe(headers=ADMIN_COLS, interactive=False, wrap=True)
ad_csv = gr.File(label="Users export", interactive=False)
with gr.Row(equal_height=False):
with gr.Column(elem_classes="cs-card"):
gr.HTML('Enable / disable account
')
ad_email = gr.Textbox(label="User email")
with gr.Row():
ad_disable = gr.Button("🚫 Disable")
ad_enable = gr.Button("✅ Enable")
with gr.Column(elem_classes="cs-card"):
gr.HTML('Reset password
')
ad_pw_email = gr.Textbox(label="User email")
ad_new_pw = gr.Textbox(label="New password", type="password")
ad_reset = gr.Button("🔑 Reset password")
with gr.Column(elem_classes="cs-card"):
gr.HTML('Delete account
')
ad_del_email = gr.Textbox(label="User email")
ad_del_confirm = gr.Textbox(label="Type the email again to confirm")
ad_delete = gr.Button("🗑️ Delete user", variant="stop")
ad_msg = gr.HTML()
gr.HTML(FOOTER)
# ---------------- Events ----------------
auth_outputs = [user_state, auth_view, app_view, user_bar, auth_msg, admin_tab, ad_stats, ad_table]
ad_out = [ad_stats, ad_table, ad_msg]
hid = dict(show_progress="hidden", api_name=False)
ad_refresh.click(lambda u, q: _admin_refresh(u, q, "Refreshed."), [user_state, ad_query], ad_out, **hid)
ad_query.submit(admin_search, [user_state, ad_query], ad_table, **hid)
ad_disable.click(lambda u, e, q: admin_set_active(u, e, q, False), [user_state, ad_email, ad_query], ad_out, **hid)
ad_enable.click(lambda u, e, q: admin_set_active(u, e, q, True), [user_state, ad_email, ad_query], ad_out, **hid)
ad_reset.click(admin_reset_pw, [user_state, ad_pw_email, ad_new_pw, ad_query], ad_out, **hid)
ad_delete.click(admin_delete, [user_state, ad_del_email, ad_del_confirm, ad_query], ad_out, **hid)
ad_export.click(admin_export, [user_state], ad_csv, **hid)
li_btn.click(do_login, [li_email, li_pw], auth_outputs, show_progress="hidden", api_name=False)
li_pw.submit(do_login, [li_email, li_pw], auth_outputs, show_progress="hidden", api_name=False)
su_btn.click(do_signup, [su_name, su_email, su_pw, su_pw2], auth_outputs, show_progress="hidden", api_name=False)
logout_btn.click(do_logout, None, auth_outputs, show_progress="hidden", api_name=False)
d_btn.click(
process_docx,
inputs=[input_file, d_temp, d_min, d_mode, user_state],
outputs=[d_out, d_dl, d_status, d_scores, d_preview],
api_name=False,
show_progress="hidden",
)
p_in.change(count_words, inputs=p_in, outputs=p_count, show_progress="hidden", queue=False, api_name=False)
p_btn.click(
process_text,
inputs=[p_in, p_temp, p_min, p_mode, user_state],
outputs=[p_out, p_file, p_dl, p_status, p_scores],
api_name=False,
show_progress="hidden",
)
demo.queue(default_concurrency_limit=2, max_size=30)
if __name__ == "__main__":
demo.launch(
show_api=False,
allowed_paths=[str(OUT_DIR)],
favicon_path=str(FAVICON_PATH) if FAVICON_PATH.exists() else None,
)