""" Caudal premium: calcular y APLICAR encode para ver 12-16 camaras en IP816. IP816 = Fast Ethernet 10/100 (~80 Mbps utiles). 16 x Extra 512 kbps ≈ 8 Mbps (cabe). 16 x Main 5MP 4 Mbps ≈ 64 Mbps en UN visor (al limite) y tumba WiFi si 7 visores lo piden. Aplica via CGI Dahua (prioridad) u ONVIF SetVideoEncoderConfiguration. Usuario/clave solo en memoria; nunca se escriben en logs ni informes. """ from __future__ import annotations import base64 import hashlib import os import re import time import urllib.parse import urllib.request from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from datetime import datetime, timezone from typing import Dict, List, Optional, Tuple from .models import Device, now_str from .onvif_rtsp import _soap_post, probe_video_encoders from .topology import _is_camera # Fast Ethernet realista: no son 100 Mbps de catalogo (cabeceras, audio, ONVIF, duplex). IP816_USEFUL_MBPS = 72.0 VISOR_OVERHEAD = 1.15 WIFI_REAL_MBPS = 25.0 PROFILES: Dict[str, Dict] = { "visor_16ch": { "label": "Visor IP816 12-16 camaras (recomendado)", "main": {"w": 1280, "h": 720, "fps": 12, "kbps": 1536, "enc": "H.265", "gop": 12}, "sub": {"w": 704, "h": 576, "fps": 10, "kbps": 512, "enc": "H.264", "gop": 10}, "note": ( "16 x Extra 512 kbps ≈ 8 Mbps al visor (cabe en 100M con holgura). " "Main 1536 kbps 720p para grabar en el PRINCIPAL. Extra en H.264 por compatibilidad IP816." ), }, "visor_wifi": { "label": "WiFi Tenda + 7 visores (agresivo)", "main": {"w": 1280, "h": 720, "fps": 10, "kbps": 1024, "enc": "H.265", "gop": 10}, "sub": {"w": 640, "h": 360, "fps": 8, "kbps": 384, "enc": "H.264", "gop": 8}, "note": ( "Para las 10 Dahua WiFi. 16 x Extra 384 kbps ≈ 6 Mbps por visor. " "7 visores x 10 WiFi en Extra ≈ 31 Mbps de aire (justo). Main alto en WiFi es imposible." ), }, "poe_only": { "label": "Solo PoE / cable (un poco mas calidad)", "main": {"w": 1920, "h": 1080, "fps": 15, "kbps": 2048, "enc": "H.265", "gop": 15}, "sub": {"w": 704, "h": 576, "fps": 12, "kbps": 640, "enc": "H.264", "gop": 12}, "note": ( "12-16 PoE en un visor con Extra 640 kbps ≈ 10 Mbps. " "No usar este perfil en las 10 WiFi Tenda." ), }, } PROFILE_AUTO = "auto" def budget_line(n_cams: int, kbps: int, n_visors: int = 1) -> Tuple[float, bool, str]: one = n_cams * kbps / 1000.0 * VISOR_OVERHEAD total = one * n_visors ok_one = one <= IP816_USEFUL_MBPS wifi_note = "" if n_visors > 1: wifi_ok = total <= WIFI_REAL_MBPS wifi_note = f"; aire WiFi 2.4 si todos tiran: {'OK/justo' if wifi_ok else 'SATURA (~25 Mbps reales)'}" mark = "OK" if ok_one else "NO" msg = ( f"{n_cams} cam x {kbps} kbps x {n_visors} visor(es) ≈ {total:.1f} Mbps " f"(1 visor 100M: {mark}{wifi_note})" ) return total, ok_one, msg def caudal_math_text() -> str: lines = [ "MATEMATICA — 12 y 16 CAMARAS EN UN IP816 (10/100 Mbps)", "-" * 54, "Puerto del visor ≈ 72 Mbps utiles (no 100 de catalogo). Cada stream x 1.15 (audio/ONVIF).", ] for n in (12, 16): lines.append(f"\n--- {n} camaras en UN visor ---") for kbps, name in ( (4096, "Main 5MP alto (tipico de fabrica)"), (2048, "Main 1080p 2 Mbps"), (1536, "Main acotado 1.5 Mbps"), (640, "Extra/Sub 640 kbps"), (512, "Extra/Sub 512 kbps"), (384, "Extra WiFi 384 kbps"), ): _, ok, msg = budget_line(n, kbps, 1) lines.append((" OK " if ok else " NO ") + msg + f" [{name}]") lines.append("\n--- 7 visores pidiendo las mismas 10 WiFi ---") for kbps, name in ((4096, "Main alto"), (1536, "Main acotado"), (512, "Extra 512"), (384, "Extra 384")): _, _, msg = budget_line(10, kbps, 7) lines.append(f" {msg} [{name}]") lines.extend( [ "", "Conclusion: 12-16 fluidas en cada TV = Extra/Sub 384-640 kbps en las camaras", "Y Extra/Sub seleccionado en HDMI de los 7 visores. El Main alto solo lo usa el principal.", "Bajar caudal en camara NO sustituye Extra/Sub en el IP816: hay que hacer las dos cosas.", ] ) return "\n".join(lines) def pick_profile(cameras: List[Device]) -> str: """WiFi (latencia alta) → perfil agresivo; si no, visor 16ch.""" if not cameras: return "visor_16ch" wifiish = sum(1 for c in cameras if (c.latency_ms or 0) >= 8) if wifiish >= max(2, len(cameras) // 3): return "visor_wifi" return "visor_16ch" def resolve_profile_key(key: str, cameras: List[Device]) -> str: if key in ("", PROFILE_AUTO, "auto"): return pick_profile(cameras) if key in PROFILES: return key return "visor_16ch" def filter_cameras(devices: List[Device]) -> List[Device]: return [d for d in devices if _is_camera(d) or d.device_type == "Camara"] def kv_lookup(kv: Dict[str, str], contains: Tuple[str, ...], not_contains: Tuple[str, ...] = ()) -> str: for k, v in kv.items(): if all(c in k for c in contains) and not any(n in k for n in not_contains): return v.strip() return "" def describe_kv_encode(kv: Dict[str, str]) -> str: def main(*parts: str) -> str: return kv_lookup(kv, ("MainFormat[0]",) + parts, ("BitRateControl", "BitRateType")) or "?" def extra(*parts: str) -> str: return kv_lookup(kv, ("ExtraFormat[0]",) + parts, ("BitRateControl", "BitRateType")) or "?" return ( f"Main {main('Width')}x{main('Height')} {main('FPS')}fps " f"{main('Video.BitRate')}kbps {main('Compression')} | " f"Extra {extra('Width')}x{extra('Height')} {extra('FPS')}fps " f"{extra('Video.BitRate')}kbps" ) def _xml_esc(s: str) -> str: return ( s.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) ) def _auth_opener(user: str, password: str, origin: str): mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm() if user: mgr.add_password(None, origin, user, password) return urllib.request.build_opener( urllib.request.HTTPDigestAuthHandler(mgr), urllib.request.HTTPBasicAuthHandler(mgr), ) def _http_call( url: str, user: str, password: str, data: Optional[bytes] = None, headers: Optional[Dict[str, str]] = None, timeout: float = 7.0, method: Optional[str] = None, ) -> str: parsed = urllib.parse.urlparse(url) origin = f"{parsed.scheme}://{parsed.netloc}" op = _auth_opener(user, password, origin) hdrs = dict(headers or {}) meth = method or ("POST" if data is not None else "GET") req = urllib.request.Request(url, data=data, headers=hdrs, method=meth) with op.open(req, timeout=timeout) as resp: return resp.read().decode("utf-8", errors="replace") def dahua_cgi_get(ip: str, user: str, password: str, timeout: float = 6.0) -> Tuple[bool, str, Dict[str, str]]: kv: Dict[str, str] = {} last_err = "" for port in (80, 8080): origin = f"http://{ip}" if port == 80 else f"http://{ip}:{port}" url = origin + "/cgi-bin/configManager.cgi?action=getConfig&name=Encode" try: text = _http_call(url, user, password, timeout=timeout) if "Error" in text and "Encode[" not in text and "Encode." not in text: last_err = text[:200] continue for line in text.splitlines(): if "=" in line: k, v = line.split("=", 1) kv[k.strip()] = v.strip() if any("BitRate" in k or "Encode" in k for k in kv): return True, text, kv last_err = text[:200] or "CGI vacio" except Exception as e: last_err = str(e) continue return False, last_err, kv def _cgi_keep_codec(current: str, fallback: str) -> str: c = (current or "").upper().replace(" ", "") if "265" in c: return "H.265" if "264" in c: return "H.264" return fallback def _set_pairs(profile_key: str, kv: Optional[Dict[str, str]], also_resolution: bool) -> List[Tuple[str, str]]: p = PROFILES[profile_key] m, s = p["main"], p["sub"] enc_m = _cgi_keep_codec(kv_lookup(kv or {}, ("MainFormat[0]", "Compression")), m["enc"]) enc_s = _cgi_keep_codec(kv_lookup(kv or {}, ("ExtraFormat[0]", "Compression")), s["enc"]) pairs: List[Tuple[str, str]] = [ ("Encode[0].MainFormat[0].VideoEnable", "true"), ("Encode[0].MainFormat[0].Video.Compression", enc_m), ("Encode[0].MainFormat[0].Video.BitRateControl", "CBR"), ("Encode[0].MainFormat[0].Video.BitRate", str(m["kbps"])), ("Encode[0].MainFormat[0].Video.FPS", str(m["fps"])), ("Encode[0].MainFormat[0].Video.GOP", str(m["gop"])), ("Encode[0].ExtraFormat[0].VideoEnable", "true"), ("Encode[0].ExtraFormat[0].Video.Compression", enc_s), ("Encode[0].ExtraFormat[0].Video.BitRateControl", "CBR"), ("Encode[0].ExtraFormat[0].Video.BitRate", str(s["kbps"])), ("Encode[0].ExtraFormat[0].Video.FPS", str(s["fps"])), ("Encode[0].ExtraFormat[0].Video.GOP", str(s["gop"])), ] if also_resolution: pairs.extend( [ ("Encode[0].MainFormat[0].Video.Width", str(m["w"])), ("Encode[0].MainFormat[0].Video.Height", str(m["h"])), ("Encode[0].ExtraFormat[0].Video.Width", str(s["w"])), ("Encode[0].ExtraFormat[0].Video.Height", str(s["h"])), ] ) return pairs def _cgi_ok(text: str) -> bool: t = (text or "").strip() if not t: return True low = t.lower() if "error" in low and "ok" not in low: return False return "ok" in low or "success" in low or "error" not in low def dahua_cgi_set( ip: str, user: str, password: str, profile_key: str, kv: Optional[Dict[str, str]] = None, also_resolution: bool = True, timeout: float = 8.0, ) -> Tuple[bool, str]: pairs = _set_pairs(profile_key, kv, also_resolution) q = ["action=setConfig"] for k, v in pairs: q.append(urllib.parse.quote(k, safe="[]().") + "=" + urllib.parse.quote(str(v), safe="")) qs = "&".join(q) last = "" for port in (80, 8080): origin = f"http://{ip}" if port == 80 else f"http://{ip}:{port}" url = origin + "/cgi-bin/configManager.cgi?" + qs try: text = _http_call(url, user, password, timeout=timeout) last = text[:400] or "CGI set sin cuerpo (posible OK)" if _cgi_ok(text): return True, last except Exception as e: last = str(e) try: body = qs.encode("ascii") text = _http_call( origin + "/cgi-bin/configManager.cgi", user, password, data=body, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=timeout, method="POST", ) last = text[:400] or "CGI POST sin cuerpo" if _cgi_ok(text): return True, last except Exception as e: last = str(e) return False, last or "CGI set fallido" def _wsse_header(user: str, password: str, digest: bool) -> str: if digest: nonce_raw = os.urandom(16) nonce_b64 = base64.b64encode(nonce_raw).decode("ascii") created = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") dig = hashlib.sha1(nonce_raw + created.encode("utf-8") + password.encode("utf-8")).digest() pwd_el = ( '' f"{base64.b64encode(dig).decode('ascii')}" '' f"{nonce_b64}" '' f"{created}" ) else: pwd_el = ( '' f"{_xml_esc(password)}" ) return ( "" '' "" f"{_xml_esc(user)}" f"{pwd_el}" "" ) def _inject_header(envelope: str, header: str) -> str: if "" in envelope or "
" in envelope: return envelope return envelope.replace("", header + "", 1) def _soap_post_auth( url: str, body: bytes, soap_action: str, timeout: float, user: str, password: str, wsse: str = "", ) -> str: xml = body.decode("utf-8", errors="replace") if wsse: xml = _inject_header(xml, wsse) body = xml.encode("utf-8") headers = { "Content-Type": "application/soap+xml; charset=utf-8", "SOAPAction": f'"{soap_action}"', } try: return _http_call(url, user, password, data=body, headers=headers, timeout=timeout, method="POST") except Exception: return _soap_post(url, body, soap_action, timeout, user, password) def _patch_limits(xml: str, w: int, h: int, fps: int, kbps: int, gop: int) -> str: def sub(tag: str, val: int, text: str) -> str: return re.sub( rf"(<(?:\w+:)?{tag}>)(\d+)()", rf"\g<1>{val}\g<3>", text, count=1, flags=re.I, ) xml = sub("Width", w, xml) xml = sub("Height", h, xml) xml = sub("FrameRateLimit", fps, xml) xml = sub("BitrateLimit", kbps, xml) xml = sub("GovLength", gop, xml) return xml def _extract_configuration_xml(raw: str) -> str: m = re.search( r"(<(?:\w+:)?VideoEncoderConfiguration\b[\s\S]*?)", raw, re.I, ) if m: return m.group(1) m = re.search( r"(<(?:\w+:)?Configuration\b[^>]*token[\s\S]*?)", raw, re.I, ) return m.group(1) if m else "" def _wrap_set_encoder(config_xml: str) -> str: """SetVideoEncoderConfiguration: el hijo debe ser Configuration (ONVIF Media).""" inner = config_xml.strip() if re.match(r"<(?:\w+:)?VideoEncoderConfiguration\b", inner, re.I): inner = re.sub( r"^<(?:\w+:)?VideoEncoderConfiguration\b", "\s*$", "", inner, count=1, flags=re.I, ) elif not re.match(r"<(?:\w+:)?Configuration\b", inner, re.I): inner = f"{inner}" return f""" {inner} true """ def _onvif_set_token( endpoint: str, token: str, w: int, h: int, fps: int, kbps: int, gop: int, encoding: str, user: str, password: str, ) -> Tuple[bool, str]: get_body = f""" {_xml_esc(token)} """.encode("utf-8") action_get = "http://www.onvif.org/ver10/media/wsdl/GetVideoEncoderConfiguration" action_set = "http://www.onvif.org/ver10/media/wsdl/SetVideoEncoderConfiguration" last = "" cfg_xml = "" for wsse_mode in ("digest", "text", ""): wsse = _wsse_header(user, password, wsse_mode == "digest") if (user and wsse_mode) else "" try: raw = _soap_post_auth(endpoint, get_body, action_get, 6.0, user, password, wsse) if "Fault" in raw or "faultcode" in raw.lower(): last = raw[:180] continue cfg_xml = _extract_configuration_xml(raw) if cfg_xml: break last = raw[:180] except Exception as e: last = str(e) if cfg_xml: patched = _patch_limits(cfg_xml, w, h, fps, kbps, gop) set_xml = _wrap_set_encoder(patched) for wsse_mode in ("digest", "text", ""): wsse = _wsse_header(user, password, wsse_mode == "digest") if (user and wsse_mode) else "" try: raw = _soap_post_auth( endpoint, set_xml.encode("utf-8"), action_set, 7.0, user, password, wsse ) if "Fault" in raw or "faultcode" in raw.lower(): last = raw[:220] continue return True, "ONVIF SetVideoEncoder OK (config existente parcheada)" except Exception as e: last = str(e) enc = "H265" if "265" in encoding.upper() else "H264" extra = ( "Main" if enc == "H264" else "Main" ) built = f""" {_xml_esc(token)} 1 {enc} {w}{h} 4 {fps} 1 {kbps} {gop} {extra} true """ try: raw = _soap_post_auth(endpoint, built.encode("utf-8"), action_set, 7.0, user, password, "") if "Fault" in raw or "faultcode" in raw.lower(): return False, (last + " | " + raw[:180]) if last else raw[:220] return True, "ONVIF SetVideoEncoder OK (config construida)" except Exception as e: return False, last or str(e) def apply_onvif_profile( ip: str, profile_key: str, user: str, password: str, also_resolution: bool = True, ) -> Tuple[bool, str]: vid = probe_video_encoders(ip, user=user, password=password) if not vid.get("ok") or not vid.get("profiles"): return False, vid.get("error") or "ONVIF sin perfiles (hace falta usuario/clave)" endpoint = vid.get("endpoint") or f"http://{ip}/onvif/media_service" profs = list(vid["profiles"]) def score(p: Dict) -> int: return (p.get("bitrate_kbps") or 0) * 10 + (p.get("width") or 0) * (p.get("height") or 0) main_p = max(profs, key=score) sub_p = min(profs, key=score) if len(profs) > 1 else None tgt = PROFILES[profile_key] msgs: List[str] = [] def tok_of(p: Dict) -> str: return (p.get("token") or p.get("profile_token") or "").strip() mw, mh = (tgt["main"]["w"], tgt["main"]["h"]) if also_resolution else ( main_p.get("width") or tgt["main"]["w"], main_p.get("height") or tgt["main"]["h"], ) tok = tok_of(main_p) if tok: ok, msg = _onvif_set_token( endpoint, tok, int(mw), int(mh), tgt["main"]["fps"], tgt["main"]["kbps"], tgt["main"]["gop"], main_p.get("encoding") or tgt["main"]["enc"], user, password, ) msgs.append(f"main token={tok}: {msg}") if not ok: return False, " | ".join(msgs) if sub_p: tok2 = tok_of(sub_p) if tok2 and tok2 != tok: sw, sh = (tgt["sub"]["w"], tgt["sub"]["h"]) if also_resolution else ( sub_p.get("width") or tgt["sub"]["w"], sub_p.get("height") or tgt["sub"]["h"], ) ok, msg = _onvif_set_token( endpoint, tok2, int(sw), int(sh), tgt["sub"]["fps"], tgt["sub"]["kbps"], tgt["sub"]["gop"], sub_p.get("encoding") or tgt["sub"]["enc"], user, password, ) msgs.append(f"sub token={tok2}: {msg}") if not ok: return False, " | ".join(msgs) return bool(msgs), " | ".join(msgs) if msgs else "ONVIF sin token de encoder" @dataclass class TuneResult: ip: str method: str ok: bool detail: str before: str = "" after: str = "" def _verify_kbps(kv: Dict[str, str], profile_key: str) -> Tuple[bool, str]: p = PROFILES[profile_key] main_br = kv_lookup(kv, ("MainFormat[0]", "Video.BitRate"), ("BitRateControl", "BitRateType")) extra_br = kv_lookup(kv, ("ExtraFormat[0]", "Video.BitRate"), ("BitRateControl", "BitRateType")) try: m_ok = abs(int(float(main_br)) - p["main"]["kbps"]) <= 64 except Exception: m_ok = False try: s_ok = abs(int(float(extra_br)) - p["sub"]["kbps"]) <= 64 except Exception: s_ok = False desc = describe_kv_encode(kv) return m_ok and s_ok, desc def apply_one_camera( ip: str, profile_key: str, user: str, password: str, dry_run: bool, also_resolution: bool = True, ) -> TuneResult: p = PROFILES[profile_key] would = ( f"Main {p['main']['kbps']} kbps {p['main']['w']}x{p['main']['h']} @{p['main']['fps']}fps CBR; " f"Extra {p['sub']['kbps']} kbps {p['sub']['w']}x{p['sub']['h']} @{p['sub']['fps']}fps" + ("" if also_resolution else " (solo bitrate/FPS, sin tocar resolucion)") ) ok_cgi, cgi_txt, kv = dahua_cgi_get(ip, user, password) if ok_cgi: before = describe_kv_encode(kv) if dry_run: return TuneResult( ip, "CGI Dahua (simulacion)", True, f"AHORA: {before}. APLICARIA: {would}.", before, ) ok, msg = dahua_cgi_set(ip, user, password, profile_key, kv, also_resolution) if not ok and also_resolution: ok, msg = dahua_cgi_set(ip, user, password, profile_key, kv, False) if ok: msg = "CGI OK en segundo intento (solo bitrate/FPS; resolucion rechazada). " + msg after = "" verified = False if ok: time.sleep(0.45) ok2, _, kv2 = dahua_cgi_get(ip, user, password) if ok2: verified, after = _verify_kbps(kv2, profile_key) if not verified: msg += " | aviso: relectura no coincide del todo (firmware puede redondear)." return TuneResult(ip, "CGI Dahua", ok, msg, before, after) vid = probe_video_encoders(ip, user=user, password=password) if vid.get("ok"): parts = [] for pr in vid.get("profiles") or []: parts.append( f"{pr.get('encoding') or '?'} {pr.get('width')}x{pr.get('height')} {pr.get('bitrate_kbps')}kbps" ) before = " ; ".join(parts) if dry_run: return TuneResult( ip, "ONVIF (simulacion)", True, f"AHORA: {before or 'perfiles leidos'}. CGI no disponible; se usaria SetVideoEncoder. APLICARIA: {would}.", before, ) ok, msg = apply_onvif_profile(ip, profile_key, user, password, also_resolution) return TuneResult(ip, "ONVIF", ok, msg, before) hint = cgi_txt[:120] if cgi_txt else "" onv = vid.get("error") or "" return TuneResult( ip, "ninguno", False, f"Sin CGI ni ONVIF Media. Pruebe usuario/clave admin de la CAMARA (no del NVR). CGI={hint} ONVIF={onv}"[:400], "", ) def apply_caudal( cameras: List[Device], profile_key: str, user: str, password: str, dry_run: bool = True, also_resolution: bool = True, workers: int = 4, ) -> str: cams = filter_cameras(cameras) key = resolve_profile_key(profile_key, cams) p = PROFILES[key] mode = "SIMULACION (no se escribe en camaras)" if dry_run else "APLICAR (cambia Encode en camaras)" lines = [ "CAUDAL PREMIUM — 12-16 CAMARAS EN IP816 SIN SATURAR", "=" * 58, f"Generado: {now_str()}", f"Perfil: {p['label']} [{key}]", p["note"], f"Modo: {mode}", f"Resolucion: {'tambien Width/Height' if also_resolution else 'solo bitrate/FPS/CBR (mas compatible)'}", f"Camaras objetivo: {len(cams)} (usuario/clave NO se guardan ni se imprimen)", "", caudal_math_text(), "", "RESULTADO POR CAMARA", "-" * 40, ] if not cams: lines.append("No hay camaras en el escaneo. ESCANEAR RED y vuelva a pulsar.") return "\n".join(lines) if not dry_run and not user: lines.append("ABORTADO: para APLICAR hace falta usuario y clave de las camaras Dahua.") return "\n".join(lines) results: List[TuneResult] = [] with ThreadPoolExecutor(max_workers=max(1, min(workers, len(cams)))) as ex: futs = { ex.submit(apply_one_camera, c.ip, key, user, password, dry_run, also_resolution): c for c in cams } for fut in as_completed(futs): try: results.append(fut.result()) except Exception as e: c = futs[fut] results.append(TuneResult(c.ip, "error", False, str(e))) def ip_key(ip: str): try: return tuple(int(x) for x in ip.split(".")) except Exception: return (0, 0, 0, 0) results.sort(key=lambda r: ip_key(r.ip)) ok_n = sum(1 for r in results if r.ok) fail_n = len(results) - ok_n for r in results: mark = "OK" if r.ok else "FAIL" lines.append(f"[{mark}] {r.ip} via {r.method}") if r.before: lines.append(f" antes: {r.before}") if r.after: lines.append(f" ahora: {r.after}") lines.append(f" {r.detail}") lines.append("") lines.append(f"Resumen: {ok_n}/{len(results)} OK, {fail_n} fallos.") lines.extend( [ "", "OBLIGATORIO EN LOS 7 IP816 DESPUES (HDMI, no lo hace la app)", "1. Preview/Encode de cada canal = Extra stream / Sub (NUNCA Main en visores).", "2. Channel Type = 16 x 5MP (si ese TV ve mas de 10 camaras).", "3. Record Disable en visores. Graba el principal.", "4. Si CGI/ONVIF FAIL: http://IP-camara → Codificacion/Video y ponga a mano los kbps del perfil.", "5. Usuario/clave = admin de la CAMARA Dahua, no los del NVR PNI.", "6. Tras aplicar: boton Fluidez/Lag y compruebe Extra ~512/384 kbps.", "7. Configurar camaras no basta si el visor sigue pidiendo Main: saturara igual.", ] ) # Cinturon: la clave no debe colarse en el informe blob = "\n".join(lines) if password and len(password) >= 4 and password in blob: blob = blob.replace(password, "********") return blob