fix(ci): add psql fallback in check_pg_query to bypass psycopg dependency in Gitea Actions

tools/verify_wbs_task_v1.py

@@ -120,59 +120,124 @@ def _build_psycopg_dsn(parsed: dict[str, str | None]) -> str:
 
 def check_pg_query(root: Path, check: dict[str, Any], dsn: str | None) -> tuple[bool, dict[str, Any]]:
     """Verify pg_query check type."""
-    if psycopg is None:
+    import os
-        return False, {"error": "psycopg not installed"}
 
     if not dsn:
         return False, {"error": "No PostgreSQL connection available"}
 
     sql = check.get("sql", "")
     expect = check.get("expect", {})
+    observed = None
+    error_msg = None
+    connected_as_host = "unknown"
 
-    try:
+    if psycopg is not None:
-        conn = psycopg.connect(dsn)
         try:
-            cursor = conn.cursor()
+            conn = psycopg.connect(dsn)
-            cursor.execute(sql)
+            try:
-            row = cursor.fetchone()
+                cursor = conn.cursor()
-            observed = row[0] if row else None
+                cursor.execute(sql)
-
+                row = cursor.fetchone()
-            # Try to coerce to numeric for comparison
+                observed = row[0] if row else None
-            if observed is not None:
+                cursor.close()
-                try:
+                conn.close()
-                    observed = float(observed)
+                connected_as_host = dsn.split("host=")[-1].split()[0] if "host=" in dsn else "unknown"
-                except (ValueError, TypeError):
+            except Exception as e:
-                    pass
+                conn.close()
-
+                error_msg = f"psycopg execution failed: {e}"
-            # Check expectations
-            passed = True
-            if "min" in expect:
-                min_val = expect["min"]
-                if observed is None or float(observed) < float(min_val):
-                    passed = False
-            if "max" in expect and passed:
-                max_val = expect["max"]
-                if observed is None or float(observed) > float(max_val):
-                    passed = False
-            if "equals" in expect and passed:
-                eq_val = expect["equals"]
-                if observed != eq_val:
-                    passed = False
-
-            cursor.close()
-            conn.close()
-
-            return passed, {
-                "sql": sql,
-                "observed": observed,
-                "expected": expect,
-                "connected_as_host": dsn.split("host=")[-1].split()[0] if "host=" in dsn else "unknown"
-            }
         except Exception as e:
-            conn.close()
+            error_msg = f"psycopg connection failed: {e}"
-            return False, {"error": str(e), "sql": sql}
+    else:
-    except Exception as e:
+        error_msg = "psycopg not installed"
-        return False, {"error": str(e), "sql": sql}
+
+    # Fallback to psql CLI if psycopg failed or is not installed
+    if observed is None:
+        import shutil
+        psql_path = shutil.which("psql")
+        if psql_path:
+            import re
+            dsn_parts = {}
+            matches = re.findall(r"(\w+)\s*=\s*(?:'([^']*)'|(\S+))", dsn)
+            for key, val1, val2 in matches:
+                dsn_parts[key] = val1 or val2
+
+            cmd = [psql_path]
+            if "host" in dsn_parts:
+                cmd.extend(["-h", dsn_parts["host"]])
+                connected_as_host = dsn_parts["host"]
+            if "port" in dsn_parts:
+                cmd.extend(["-p", dsn_parts["port"]])
+            if "user" in dsn_parts:
+                cmd.extend(["-U", dsn_parts["user"]])
+            if "dbname" in dsn_parts:
+                cmd.extend(["-d", dsn_parts["dbname"]])
+            cmd.extend(["-t", "-A", "-c", sql])
+
+            env = dict(os.environ)
+            if "password" in dsn_parts:
+                env["PGPASSWORD"] = dsn_parts["password"]
+
+            try:
+                res = subprocess.run(cmd, env=env, capture_output=True, text=True, check=True)
+                stdout_val = res.stdout.strip()
+                if stdout_val:
+                    observed = stdout_val
+                error_msg = None
+            except Exception as e:
+                stderr_msg = e.stderr if hasattr(e, "stderr") else ""
+                error_msg = f"psql fallback failed: {e}. stderr: {stderr_msg}. (Previous: {error_msg})"
+        else:
+            error_msg = f"psql CLI not found. (Previous: {error_msg})"
+
+    if error_msg is not None:
+        return False, {"error": error_msg, "sql": sql}
+
+    # Try to coerce to numeric for comparison
+    if observed is not None:
+        try:
+            observed_float = float(observed)
+            if observed_float.is_integer():
+                observed = int(observed_float)
+            else:
+                observed = observed_float
+        except (ValueError, TypeError):
+            pass
+
+    # Check expectations
+    passed = True
+    if "min" in expect:
+        min_val = expect["min"]
+        try:
+            if observed is None or float(observed) < float(min_val):
+                passed = False
+        except (ValueError, TypeError):
+            passed = False
+    if "max" in expect and passed:
+        max_val = expect["max"]
+        try:
+            if observed is None or float(observed) > float(max_val):
+                passed = False
+        except (ValueError, TypeError):
+            passed = False
+    if "equals" in expect and passed:
+        eq_val = expect["equals"]
+        try:
+            if isinstance(eq_val, (int, float)) or (isinstance(eq_val, str) and eq_val.replace(".","",1).isdigit()):
+                if float(observed) != float(eq_val):
+                    passed = False
+            else:
+                if str(observed) != str(eq_val):
+                    passed = False
+        except (ValueError, TypeError):
+            if str(observed) != str(eq_val):
+                passed = False
+
+    return passed, {
+        "sql": sql,
+        "observed": observed,
+        "expected": expect,
+        "connected_as_host": connected_as_host
+    }
 
 
 def check_log_pattern(root: Path, check: dict[str, Any]) -> tuple[bool, dict[str, Any]]: