Gestion Citas Peluqueria
Aplicación Gestion Citas Peluqueria para gestión y administración local.
Registros Existentes
| ID |
Cliente | Estado_la_cita |
Acciones |
"""
class DataStore:
"""Manages persistent JSON storage for Turno entities."""
def __init__(self, filepath: str = DATA_FILE):
self.filepath = filepath
self._ensure_storage()
def _ensure_storage(self) -> None:
if not os.path.exists(self.filepath):
initial_data = {"records": [], "created_at": time.time(), "project": "Gestion Citas Peluqueria"}
with open(self.filepath, "w", encoding="utf-8") as f:
json.dump(initial_data, f, indent=2, ensure_ascii=False)
def load_all(self) -> List[Dict[str, Any]]:
self._ensure_storage()
try:
with open(self.filepath, "r", encoding="utf-8") as f:
data = json.load(f)
return data.get("records", [])
except Exception:
return []
def save_all(self, records: List[Dict[str, Any]]) -> None:
payload = {"records": records, "updated_at": time.time(), "project": "Gestion Citas Peluqueria"}
with open(self.filepath, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False)
def add(self, record: Dict[str, Any]) -> Dict[str, Any]:
records = self.load_all()
if "id" not in record or not record["id"]:
record["id"] = f"REC-{int(time.time()*1000) % 100000:05d}"
if "created_at" not in record:
record["created_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
if "estado_la_cita" not in record:
record["estado_la_cita"] = "PENDIENTE"
records.append(record)
self.save_all(records)
return record
def find_by_id(self, rec_id: str) -> Optional[Dict[str, Any]]:
for r in self.load_all():
if str(r.get("id")) == str(rec_id):
return r
return None
def update(self, rec_id: str, updates: Dict[str, Any]) -> bool:
records = self.load_all()
found = False
for r in records:
if str(r.get("id")) == str(rec_id):
r.update(updates)
r["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
found = True
break
if found:
self.save_all(records)
return found
def delete(self, rec_id: str) -> bool:
records = self.load_all()
initial_len = len(records)
filtered = [r for r in records if str(r.get("id")) != str(rec_id)]
if len(filtered) < initial_len:
self.save_all(filtered)
return True
return False
class TurnoService:
"""Business logic service for Turno operations."""
def __init__(self, store: Optional[DataStore] = None):
self.store = store or DataStore()
def register(self, **kwargs) -> Dict[str, Any]:
"""Registers a new Turno with all required fields."""
rec = {k: v for k, v in kwargs.items() if v is not None}
return self.store.add(rec)
def list_all(self, estado: Optional[str] = None) -> List[Dict[str, Any]]:
"""Lists records, optionally filtered by status."""
recs = self.store.load_all()
if estado:
recs = [r for r in recs if str(r.get("estado_la_cita", "")).upper() == estado.upper()]
return recs
def change_status(self, rec_id: str, new_status: str) -> bool:
"""Updates record status."""
return self.store.update(rec_id, {"estado_la_cita": new_status.upper()})
def remove(self, rec_id: str) -> bool:
"""Removes a record."""
return self.store.delete(rec_id)
def summary(self) -> Dict[str, Any]:
"""Returns statistics and summary counts."""
recs = self.store.load_all()
by_status = {}
for r in recs:
st = str(r.get("estado_la_cita", "DESCONOCIDO")).upper()
by_status[st] = by_status.get(st, 0) + 1
return {
"total": len(recs),
"by_status": by_status,
"status_field": "estado_la_cita",
"project": "Gestion Citas Peluqueria"
}
class WebAppHandler(BaseHTTPRequestHandler):
"""Embedded HTTP handler for local web interaction and REST API."""
service = TurnoService()
def _set_headers(self, status: int = 200, content_type: str = "application/json", content_length: Optional[int] = None):
self.send_response(status)
self.send_header("Content-Type", content_type)
if content_length is not None:
self.send_header("Content-Length", str(content_length))
self.send_header("Connection", "close")
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.end_headers()
def do_OPTIONS(self):
self._set_headers(200, content_length=0)
def do_GET(self):
if self.path == "/" or self.path == "/index.html":
body = HTML_PAGE.encode("utf-8")
self._set_headers(200, "text/html; charset=utf-8", len(body))
self.wfile.write(body)
elif self.path == "/api/records":
recs = self.service.list_all()
body = json.dumps(recs, ensure_ascii=False).encode("utf-8")
self._set_headers(200, "application/json", len(body))
self.wfile.write(body)
elif self.path == "/api/summary":
summ = self.service.summary()
body = json.dumps(summ, ensure_ascii=False).encode("utf-8")
self._set_headers(200, "application/json", len(body))
self.wfile.write(body)
else:
body = b'{"error": "Not found"}'
self._set_headers(404, "application/json", len(body))
self.wfile.write(body)
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length).decode("utf-8") if length > 0 else "{}"
try:
data = json.loads(body)
except Exception:
data = {}
if self.path == "/api/records":
rec = self.service.register(**data)
resp_bytes = json.dumps(rec, ensure_ascii=False).encode("utf-8")
self._set_headers(201, "application/json", len(resp_bytes))
self.wfile.write(resp_bytes)
elif self.path == "/api/records/status":
rec_id = data.get("id")
new_st = str(data.get("status") or data.get("new_value") or data.get("estado_la_cita") or data.get("estado", "COMPLETADO"))
ok = self.service.change_status(rec_id, new_st) if rec_id else False
resp_bytes = json.dumps({"success": ok, "id": rec_id, "status": new_st, "field": "estado_la_cita"}, ensure_ascii=False).encode("utf-8")
self._set_headers(200 if ok else 400, "application/json", len(resp_bytes))
self.wfile.write(resp_bytes)
elif self.path == "/api/records/delete":
rec_id = data.get("id")
ok = self.service.remove(rec_id) if rec_id else False
resp_bytes = json.dumps({"success": ok, "id": rec_id}, ensure_ascii=False).encode("utf-8")
self._set_headers(200 if ok else 400, "application/json", len(resp_bytes))
self.wfile.write(resp_bytes)
else:
body = b'{"error": "Endpoint not found"}'
self._set_headers(404, "application/json", len(body))
self.wfile.write(body)
def run_web_server(port: int = 8000, host: str = "127.0.0.1") -> None:
server_address = (host, port)
try:
httpd = ThreadingHTTPServer(server_address, WebAppHandler)
httpd.daemon_threads = True
print(f"[WEB SERVER] Running at http://{host}:{port}/")
httpd.serve_forever()
except OSError as e:
print(f"Error starting web server on {host}:{port}: {e}")
sys.exit(1)
def run_tests() -> int:
"""Self-test verification suite validating exact required fields and persistence."""
print("=== RUNNING SELF-TEST SUITE FOR Gestion Citas Peluqueria ===")
test_db = os.path.join(os.path.dirname(os.path.abspath(__file__)), "test_data.json")
if os.path.exists(test_db):
os.remove(test_db)
store = DataStore(test_db)
service = TurnoService(store)
# Test 1: Add record with all required semantic fields
sample_data = {
"cliente": 'Ana Torres',
"estado_la_cita": 'PENDIENTE'
}
rec1 = service.register(**sample_data)
assert rec1["id"] is not None, "Error: record ID should not be None"
print(f"[PASS] Register created: {rec1['id']}")
# Test 2: Verify physical persistence on disk and exact field values
item = store.find_by_id(rec1["id"])
assert item is not None, "Error: record not found in persistent store"
assert "cliente" in item, "Error: field cliente missing from storage"
assert item["cliente"] == 'Ana Torres', f"Error: cliente value mismatch"
print(f"[PASS] Field verified: cliente == {item['cliente']}")
assert "estado_la_cita" in item, "Error: field estado_la_cita missing from storage"
assert item["estado_la_cita"] == 'PENDIENTE', f"Error: estado_la_cita value mismatch"
print(f"[PASS] Field verified: estado_la_cita == {item['estado_la_cita']}")
# Test 3: List records
all_recs = service.list_all()
assert len(all_recs) == 1, f"Error: expected 1 record, got {len(all_recs)}"
assert all_recs[0]["id"] == rec1["id"]
print(f"[PASS] List retrieved: {len(all_recs)} record(s)")
# Test 4: Update status
updated = service.change_status(rec1["id"], "COMPLETADO")
assert updated is True, "Error: status update should return True"
item = store.find_by_id(rec1["id"])
assert item.get("estado_la_cita") == "COMPLETADO", f"Error: status did not match COMPLETADO, found {item.get('estado_la_cita')}"
print("[PASS] Status updated to COMPLETADO")
# Test 5: Summary
summ = service.summary()
assert summ["total"] == 1, "Error: summary total should be 1"
print(f"[PASS] Summary: {summ}")
# Test 6: Delete & consistent storage
deleted = service.remove(rec1["id"])
assert deleted is True, "Error: delete should return True"
assert len(service.list_all()) == 0, "Error: database should be empty after delete"
print("[PASS] Record deleted successfully")
# Cleanup test db
if os.path.exists(test_db):
os.remove(test_db)
print("=== ALL VERIFICATION TESTS PASSED (EXIT CODE 0) ===")
return 0
def format_table(records: List[Dict[str, Any]]) -> str:
if not records:
return "No hay registros disponibles."
keys = ["id"] + ['cliente', 'estado_la_cita'] + ["created_at"]
# De-duplicate while preserving order
ordered_keys = []
for k in keys:
if k not in ordered_keys:
ordered_keys.append(k)
col_widths = {k: max(len(k), max(len(str(r.get(k, ""))) for r in records)) for k in ordered_keys}
header = " | ".join(f"{k.upper():<{col_widths[k]}}" for k in ordered_keys)
divider = "-+-".join("-" * col_widths[k] for k in ordered_keys)
rows = []
for r in records:
row = " | ".join(f"{str(r.get(k, '')):<{col_widths[k]}}" for k in ordered_keys)
rows.append(row)
return f"{header}\n{divider}\n" + "\n".join(rows)
def main():
parser = argparse.ArgumentParser(description="Gestion Citas Peluqueria - Aplicación Gestion Citas Peluqueria para gestión y administración local.")
parser.add_argument("--test", action="store_true", help="Ejecuta suite de autoverificacion")
parser.add_argument("--serve", action="store_true", help="Inicia servidor web local interactivo")
parser.add_argument("--host", type=str, default="127.0.0.1", help="Host para servidor web")
parser.add_argument("--port", type=int, default=8000, help="Puerto para servidor web")
parser.add_argument("--list", action="store_true", help="Lista todos los registros")
parser.add_argument("--status", type=str, default=None, help="Filtra por estado")
parser.add_argument("--add", action="store_true", help="Agrega un nuevo registro")
parser.add_argument("--cliente", type=str, default=None, help="Cliente")
parser.add_argument("--estado-la-cita", type=str, default=None, help="Estado_la_cita")
parser.add_argument("--set-status", nargs=2, metavar=("ID", "NUEVO_ESTADO"), help="Actualiza estado de un registro")
parser.add_argument("--delete", type=str, metavar="ID", help="Elimina un registro por ID")
parser.add_argument("--summary", action="store_true", help="Muestra resumen de registros")
args = parser.parse_args()
if args.test:
sys.exit(run_tests())
service = TurnoService()
if args.serve:
run_web_server(port=args.port, host=args.host)
return
if args.add:
data = {}
val = getattr(args, "cliente", None)
data["cliente"] = val if val is not None else 'Ana Torres'
val = getattr(args, "estado_la_cita", None)
data["estado_la_cita"] = val if val is not None else 'PENDIENTE'
rec = service.register(**data)
print(f"Registro agregado exitosamente con ID: {rec['id']}")
sys.exit(0)
if args.set_status:
rec_id, st = args.set_status
ok = service.change_status(rec_id, st)
if ok:
print(f"Registro {rec_id} actualizado a estado '{st.upper()}'.")
else:
print(f"Error: no se encontro registro con ID {rec_id}.")
sys.exit(0 if ok else 1)
if args.delete:
ok = service.remove(args.delete)
if ok:
print(f"Registro {args.delete} eliminado correctamente.")
else:
print(f"Error: no se encontro registro {args.delete}.")
sys.exit(0 if ok else 1)
if args.summary:
s = service.summary()
print(f"=== Resumen de {s['project']} ===")
print(f"Total registros: {s['total']}")
for k, v in s['by_status'].items():
print(f" - {k}: {v}")
sys.exit(0)
if args.list:
records = service.list_all(estado=args.status)
print(f"=== {'Gestion Citas Peluqueria'.upper()} ===")
print(format_table(records))
sys.exit(0)
# Default action: start web server if interactive web app, else display table
if True:
print(f"Iniciando panel web interactivo para Gestion Citas Peluqueria...")
run_web_server(port=args.port, host=args.host)
else:
records = service.list_all(estado=args.status)
print(f"=== {'Gestion Citas Peluqueria'.upper()} ===")
print(format_table(records))
if __name__ == "__main__":
main()