forked from finos/open-resource-broker
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathquality_check.py
More file actions
executable file
·949 lines (788 loc) · 34.8 KB
/
Copy pathquality_check.py
File metadata and controls
executable file
·949 lines (788 loc) · 34.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
#!/usr/bin/env python3
"""
Quality Check Tool
A quality checker that enforces coding standards:
- No emojis in code or comments
- No unprofessional language
- No hyperbolic marketing terms
- Appropriate docstring coverage and format
- Consistent naming conventions
- No unused imports or commented code
- README files are up-to-date
Usage:
python dev-tools/scripts/quality_check.py [--fix] [--strict] [--files FILE1 FILE2...]
Options:
--fix Attempt to automatically fix issues where possible
--strict Exit with error code on any violation (for CI)
--files Specific files to check (default: git modified files)
"""
import argparse
import ast
import logging
import os
import re
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Optional
# Setup logging
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger(__name__)
try:
import pathspec
except ImportError:
pathspec = None
# --- Configuration ---
# Emoji detection pattern
EMOJI_PATTERN = re.compile(
"["
"\U0001f600-\U0001f64f" # emoticons
"\U0001f300-\U0001f5ff" # symbols & pictographs
"\U0001f680-\U0001f6ff" # transport & map
"\U0001f1e0-\U0001f1ff" # flags
"\U00002700-\U000027bf" # dingbats (fixed range)
"]+",
flags=re.UNICODE,
)
# Legitimate technical characters that should be allowed
ALLOWED_TECHNICAL_CHARS = {
"├",
"└",
"│",
"─", # Box drawing characters for tree structures
"▪",
"▫",
"■",
"□", # Simple geometric shapes for bullets
"→",
"←",
"↑",
"↓", # Basic arrows for flow diagrams
}
# Pre-compile regex patterns for better performance
UNPROFESSIONAL_PATTERNS = {
re.compile(r"\bawesome\b", re.IGNORECASE): 'Use "excellent" or specific technical terms',
re.compile(r"\brock\b", re.IGNORECASE): 'Use "implement" or "execute"',
re.compile(r"\bcool\b", re.IGNORECASE): 'Use "effective" or specific benefits',
re.compile(r"\bsweet\b", re.IGNORECASE): 'Use "beneficial" or specific advantages',
re.compile(r"\bsick\b", re.IGNORECASE): 'Use "impressive" or specific technical terms',
re.compile(r"\bepic\b", re.IGNORECASE): 'Use "comprehensive" or specific scope',
re.compile(r"\binsane\b", re.IGNORECASE): 'Use "significant" or specific metrics',
re.compile(r"\bcrazy\b", re.IGNORECASE): 'Use "substantial" or specific details',
}
# Hyperbolic marketing terms
HYPERBOLIC_PATTERNS = {
re.compile(r"\benhanced\b", re.IGNORECASE): 'Use "improved" only when factually accurate',
re.compile(r"\bunified\b", re.IGNORECASE): 'Use "integrated" or "consolidated"',
re.compile(
r"\bproper\b(?!\s*(?:ty|ties))", re.IGNORECASE
): 'Use specific terms: "correct", "appropriate", "compliant", "structured", "domain-driven"',
re.compile(r"\bmodern\b", re.IGNORECASE): 'Use "current" or "updated"',
re.compile(r"\bcutting-edge\b", re.IGNORECASE): 'Use "current industry standard"',
re.compile(r"\brevolutionary\b", re.IGNORECASE): 'Use "significant improvement"',
re.compile(r"\bnext-generation\b", re.IGNORECASE): "Use specific technology names",
re.compile(r"\bstate-of-the-art\b", re.IGNORECASE): 'Use "current best practice"',
}
# Implementation detail terms that should be removed from production code
IMPLEMENTATION_DETAIL_PATTERNS = {
re.compile(r"\bphase\s+\d+\b", re.IGNORECASE): "Remove implementation phase references",
re.compile(r"\bphase\s+[a-z]+\b", re.IGNORECASE): "Remove implementation phase references",
re.compile(r"\bmigrated\s+from\b", re.IGNORECASE): "Remove migration history references",
re.compile(r"\bmigrating\s+to\b", re.IGNORECASE): "Remove migration process references",
re.compile(r"\bthis\s+instead\s+of\s+that\b", re.IGNORECASE): "Remove comparison references",
re.compile(r"\bold\s+implementation\b", re.IGNORECASE): "Remove old implementation references",
re.compile(r"\bnew\s+implementation\b", re.IGNORECASE): "Remove new implementation references",
re.compile(r"\blegacy\s+code\b", re.IGNORECASE): "Remove legacy code references",
re.compile(r"\btemporary\s+fix\b", re.IGNORECASE): "Remove temporary implementation references",
re.compile(r"\btodo\s*:\b", re.IGNORECASE): "Remove TODO comments from production code",
re.compile(r"\bfixme\s*:\b", re.IGNORECASE): "Remove FIXME comments from production code",
re.compile(r"\bhack\s*:\b", re.IGNORECASE): "Remove HACK comments from production code",
re.compile(r"\bworkaround\s+for\b", re.IGNORECASE): "Remove workaround references",
re.compile(r"\bquick\s+fix\b", re.IGNORECASE): "Remove quick fix references",
re.compile(r"\bstep\s+\d+\b", re.IGNORECASE): "Remove step-by-step implementation references",
re.compile(r"\btest\s+\d+\b", re.IGNORECASE): "Remove test numbering from production code",
}
# Legitimate version references that should be excluded
VERSION_EXCLUSIONS = {
"CODE_OF_CONDUCT.md", # Contributor Covenant version references
"LICENSE", # License version references
"pyproject.toml", # Package version references
"version.py", # Version files
"versions.json", # Version configuration files
}
# File extensions to check
CODE_EXTENSIONS = {".py"}
DOC_EXTENSIONS = {".md", ".rst", ".txt"}
CONFIG_EXTENSIONS = {".yaml", ".yml", ".json", ".toml"}
ALL_EXTENSIONS = CODE_EXTENSIONS | DOC_EXTENSIONS | CONFIG_EXTENSIONS
# --- Violation Classes ---
class Violation:
"""Base class for quality check violations."""
def __init__(self, file_path: str, line_num: int, content: str, message: str):
self.file_path = file_path
self.line_num = line_num
self.content = content
self.message = message
def __str__(self) -> str:
return f"{self.file_path}:{self.line_num}: {self.message}\n {self.content}"
def can_autofix(self) -> bool:
"""Whether this violation can be automatically fixed."""
return False
def autofix(self) -> Optional[str]:
"""Return fixed content if possible, None otherwise."""
return None
class EmojiViolation(Violation):
"""Emoji found in code or comments."""
def __init__(self, file_path: str, line_num: int, content: str):
super().__init__(
file_path,
line_num,
content,
"Contains emoji - not allowed in professional code",
)
class UnprofessionalLanguageViolation(Violation):
"""Unprofessional language found in code or comments."""
def __init__(self, file_path: str, line_num: int, content: str, term: str, suggestion: str):
super().__init__(
file_path, line_num, content, f"Unprofessional term '{term}' - {suggestion}"
)
self.term = term
self.suggestion = suggestion
class HyperbolicTermViolation(Violation):
"""Hyperbolic marketing term found in code or comments."""
def __init__(self, file_path: str, line_num: int, content: str, term: str, suggestion: str):
super().__init__(file_path, line_num, content, f"Hyperbolic term '{term}' - {suggestion}")
self.term = term
self.suggestion = suggestion
class ImplementationDetailViolation(Violation):
"""Implementation detail term found in production code."""
def __init__(self, file_path: str, line_num: int, content: str, term: str, suggestion: str):
super().__init__(
file_path,
line_num,
content,
f"Implementation detail '{term}' - {suggestion}",
)
self.term = term
self.suggestion = suggestion
class MissingDocstringViolation(Violation):
"""Missing docstring in class, function, or module."""
def __init__(self, file_path: str, line_num: int, element_type: str, element_name: str):
super().__init__(
file_path,
line_num,
f"{element_type} {element_name}",
f"Missing docstring for {element_type} {element_name}",
)
self.element_type = element_type
self.element_name = element_name
class DocstringFormatViolation(Violation):
"""Docstring doesn't follow the required format."""
def __init__(
self,
file_path: str,
line_num: int,
element_type: str,
element_name: str,
issue: str,
):
super().__init__(
file_path,
line_num,
f"{element_type} {element_name}",
f"Docstring format issue in {element_type} {element_name}: {issue}",
)
class UnusedImportViolation(Violation):
"""Unused import found in code."""
def __init__(self, file_path: str, line_num: int, message: str):
super().__init__(file_path, line_num, "", f"Unused imports detected: {message}")
def can_autofix(self) -> bool:
return True
class CommentedCodeViolation(Violation):
"""Commented-out code found."""
def __init__(self, file_path: str, line_num: int, content: str):
super().__init__(file_path, line_num, content, "Commented-out code should be removed")
class DebugStatementViolation(Violation):
"""Debug print/logging statement found."""
def __init__(self, file_path: str, line_num: int, content: str):
super().__init__(
file_path,
line_num,
content,
"Debug print/logging statement should be removed",
)
def can_autofix(self) -> bool:
return True
class UnjustifiedGetAttrViolation(Violation):
"""getattr call without a justifying ``# getattr`` comment."""
def __init__(self, file_path: str, line_num: int, content: str):
super().__init__(
file_path,
line_num,
content,
"getattr should only be used when absolutely needed; "
"justify with a comment starting with '# getattr' on the same "
"or a preceding line if it is needed, remove getattr if not. Strongly prefer"
"removing, use web search to find the real API for external SDK calls. Do not replace"
"with an equivalent pattern that avoids getattr but is still the same anti-pattern.",
)
# --- Checker Classes ---
class FileChecker:
"""Base class for file-based checkers."""
def check_file(self, file_path: str) -> list[Violation]:
"""Check a file for violations."""
violations = []
try:
with open(file_path, encoding="utf-8") as f:
content = f.read()
violations.extend(self.check_content(file_path, content))
except UnicodeDecodeError:
# Skip binary files
pass
except Exception as e:
logger.error(f"Error checking {file_path}: {e}")
return violations
def check_content(self, file_path: str, content: str) -> list[Violation]:
"""Check file content for violations."""
return []
class EmojiChecker(FileChecker):
"""Check for emojis in files."""
def check_content(self, file_path: str, content: str) -> list[Violation]:
violations = []
for line_num, line in enumerate(content.splitlines(), 1):
matches = EMOJI_PATTERN.findall(line)
for match in matches:
# Check if all characters in the match are allowed technical chars
if not all(char in ALLOWED_TECHNICAL_CHARS for char in match):
violations.append(EmojiViolation(file_path, line_num, line.strip()))
break # Only report once per line
return violations
class LanguageChecker(FileChecker):
"""Check for unprofessional language and hyperbolic terms."""
def check_content(self, file_path: str, content: str) -> list[Violation]:
violations = []
for line_num, line in enumerate(content.splitlines(), 1):
# Check for unprofessional language
for pattern, suggestion in UNPROFESSIONAL_PATTERNS.items():
matches = pattern.finditer(line)
for match in matches:
term = match.group(0)
violations.append(
UnprofessionalLanguageViolation(
file_path, line_num, line.strip(), term, suggestion
)
)
# Check for hyperbolic terms (skip CHANGELOG.md as it contains historical commit messages)
if not file_path.endswith("CHANGELOG.md"):
for pattern, suggestion in HYPERBOLIC_PATTERNS.items():
matches = pattern.finditer(line)
for match in matches:
term = match.group(0)
violations.append(
HyperbolicTermViolation(
file_path, line_num, line.strip(), term, suggestion
)
)
# Check for implementation detail terms
for pattern, suggestion in IMPLEMENTATION_DETAIL_PATTERNS.items():
matches = pattern.finditer(line)
for match in matches:
term = match.group(0)
# Skip version references in legitimate files
if "version" in term.lower() and any(
excluded in file_path for excluded in VERSION_EXCLUSIONS
):
continue
violations.append(
ImplementationDetailViolation(
file_path, line_num, line.strip(), term, suggestion
)
)
return violations
class DocstringChecker(FileChecker):
"""Check for docstring coverage and format."""
def check_content(self, file_path: str, content: str) -> list[Violation]:
if not file_path.endswith(".py"):
return []
# Skip docstring checks for test files
if "/test" in file_path or file_path.startswith("test"):
return []
violations = []
try:
tree = ast.parse(content)
# Check module docstring (skip empty __init__.py files)
if not ast.get_docstring(tree):
# Skip empty __init__.py files - they're just package markers
if not (Path(file_path).name == "__init__.py" and len(content.strip()) == 0):
violations.append(
MissingDocstringViolation(file_path, 1, "module", Path(file_path).name)
)
# Check classes and functions
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
if not ast.get_docstring(node):
violations.append(
MissingDocstringViolation(file_path, node.lineno, "class", node.name)
)
elif isinstance(node, ast.FunctionDef):
# Skip private methods (starting with _)
if not node.name.startswith("_") or node.name == "__init__":
if not ast.get_docstring(node):
# Special handling for __init__ methods - use fast check
if node.name == "__init__" and len(node.body) <= 8:
# Quick heuristic: if small body, likely simple
if self._is_simple_init_fast(node):
continue
violations.append(
MissingDocstringViolation(
file_path, node.lineno, "function", node.name
)
)
except SyntaxError:
# Skip files with syntax errors
pass
return violations
def _is_simple_init_fast(self, node: ast.FunctionDef) -> bool:
"""Fast check if __init__ method is simple (only parameter assignment)."""
# Only check small methods
if len(node.body) > 8:
return False
# Quick pattern check: only assignments and super() calls
for stmt in node.body:
if isinstance(stmt, ast.Assign):
# Must be self.x = y pattern
if not (
len(stmt.targets) == 1
and isinstance(stmt.targets[0], ast.Attribute)
and isinstance(stmt.targets[0].value, ast.Name)
and stmt.targets[0].value.id == "self"
):
return False
elif isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Call):
# Allow super().__init__() only
continue
else:
return False
return True
class ImportChecker(FileChecker):
"""Check for unused imports using autoflake."""
def check_content(self, file_path: str, content: str) -> list[Violation]:
if not file_path.endswith(".py"):
return []
violations = []
try:
import subprocess
# Run autoflake in check mode
result = subprocess.run(
[
"autoflake",
"--check",
"--remove-all-unused-imports",
"--remove-unused-variables",
file_path,
],
check=False,
capture_output=True,
text=True,
cwd=".",
)
# If autoflake found issues, it returns non-zero exit code
if result.returncode != 0:
violations.append(
UnusedImportViolation(file_path, 1, "Run 'make format' to fix automatically")
)
except (subprocess.SubprocessError, FileNotFoundError):
# Skip if autoflake not available
pass
return violations
class CommentChecker(FileChecker):
"""Check for TODO/FIXME comments without tickets and commented code.
Supports noqa suppressions for commented code:
- Line-level: # def function(): # noqa:COMMENTED
- Section-level:
# noqa:COMMENTED section-start
# def function():
# class MyClass:
# noqa:COMMENTED section-end
"""
def check_content(self, file_path: str, content: str) -> list[Violation]:
violations = []
# Regex for commented code (simple heuristic)
code_pattern = re.compile(
r"^\s*#\s*(def|class|if|for|while|try|except|return|import|from)\s"
)
# Regex for debug prints (only catch print statements, not logger)
debug_pattern = re.compile(r"^\s*print\(")
# Skip debug print checks for test files and markdown files
is_test_file = "/test" in file_path or file_path.startswith("test")
is_markdown_file = file_path.endswith(".md")
# Track section-level suppressions and string contexts
commented_code_suppressed = False
in_multiline_string = False
string_delimiter = None
for line_num, line in enumerate(content.splitlines(), 1):
# Track multiline strings (docstrings and regular strings)
stripped = line.strip()
# Check for start/end of multiline strings
if not in_multiline_string:
if stripped.startswith('"""') or stripped.startswith("'''"):
string_delimiter = stripped[:3]
if not (stripped.endswith(string_delimiter) and len(stripped) > 3):
in_multiline_string = True
elif stripped.startswith('r"""') or stripped.startswith("r'''"):
string_delimiter = stripped[1:4]
if not (stripped.endswith(string_delimiter) and len(stripped) > 4):
in_multiline_string = True
elif stripped.endswith(string_delimiter):
in_multiline_string = False
string_delimiter = None
# Skip checks if we're inside a multiline string
if in_multiline_string:
continue
# Check for section-level suppression controls
if "# noqa:COMMENTED section-start" in line:
commented_code_suppressed = True
continue
elif "# noqa:COMMENTED section-end" in line:
commented_code_suppressed = False
continue
# Check for commented code
if code_pattern.search(line):
# Skip if suppressed by section-level or line-level noqa
if (
commented_code_suppressed
or "noqa" in line.lower()
or "noqa:commented" in line.lower()
):
continue
violations.append(CommentedCodeViolation(file_path, line_num, line.strip()))
# Check for debug statements (skip for test files and markdown files)
if (
not is_test_file
and not is_markdown_file
and debug_pattern.search(line)
and "DEBUG" not in line.upper()
and "noqa" not in line.lower()
):
violations.append(DebugStatementViolation(file_path, line_num, line.strip()))
return violations
class GetAttrChecker(FileChecker):
"""Enforce justified ``getattr`` use in Azure and GCP provider code.
Every ``getattr(...)`` call in ``src/orb/providers/azure/`` and
``src/orb/providers/gcp/`` must have a comment starting with ``# getattr``
on the same line, on a preceding line within the same scope, or in the
enclosing function/method docstring. This keeps defensive SDK access
intentional and documented at the provider boundary.
"""
_PROVIDER_PREFIXES = (
os.path.join("src", "orb", "providers", "azure"),
os.path.join("src", "orb", "providers", "gcp"),
)
_GETATTR_CALL = re.compile(r"\bgetattr\s*\(")
_JUSTIFICATION_COMMENT = re.compile(r"#\s*getattr\b")
_JUSTIFICATION_DOCSTRING = re.compile(r"\bgetattr\b")
_SCOPE_BOUNDARY = re.compile(r"^\s*(def |class )")
_MAX_LOOKBACK = 30
def check_content(self, file_path: str, content: str) -> list[Violation]:
if not file_path.endswith(".py"):
return []
normalised = os.path.normpath(file_path)
if not any(prefix in normalised for prefix in self._PROVIDER_PREFIXES):
return []
violations: list[Violation] = []
lines = content.splitlines()
# Pre-compute the docstring that covers each function/method body so
# we can check it cheaply per getattr line.
scope_docstrings = self._build_scope_docstring_map(lines)
for idx, line in enumerate(lines):
if not self._GETATTR_CALL.search(line):
continue
if self._is_justified(lines, idx, scope_docstrings):
continue
violations.append(
UnjustifiedGetAttrViolation(file_path, idx + 1, line.strip())
)
return violations
def _is_justified(
self,
lines: list[str],
idx: int,
scope_docstrings: dict[int, str],
) -> bool:
"""Return True if the getattr at *idx* has a visible justification."""
line = lines[idx]
# 1. Same-line comment.
if self._JUSTIFICATION_COMMENT.search(line):
return True
# 2. Scan backward within the same scope for a ``# getattr`` comment.
for back in range(1, self._MAX_LOOKBACK + 1):
prev_idx = idx - back
if prev_idx < 0:
break
prev = lines[prev_idx]
if self._JUSTIFICATION_COMMENT.search(prev):
return True
if self._SCOPE_BOUNDARY.search(prev):
break
# 3. Enclosing function/method docstring mentions ``getattr``.
for scope_start, docstring in scope_docstrings.items():
if scope_start < idx and self._JUSTIFICATION_DOCSTRING.search(docstring):
# Check that *idx* is inside this scope (no newer scope in
# between).
next_scope = min(
(s for s in scope_docstrings if s > scope_start),
default=len(lines),
)
if idx < next_scope:
return True
return False
@staticmethod
def _build_scope_docstring_map(lines: list[str]) -> dict[int, str]:
"""Map ``def``/``class`` line indices to their docstrings (if any)."""
scope_boundary = re.compile(r"^\s*(def |class )")
result: dict[int, str] = {}
for idx, line in enumerate(lines):
if not scope_boundary.search(line):
continue
# Look for a docstring on the next non-blank line.
doc_start = idx + 1
while doc_start < len(lines) and not lines[doc_start].strip():
doc_start += 1
if doc_start >= len(lines):
continue
first = lines[doc_start].strip()
if not (first.startswith('"""') or first.startswith("'''")):
continue
quote = first[:3]
if first.count(quote) >= 2:
# Single-line docstring.
result[idx] = first
else:
# Multi-line docstring — collect until closing quotes.
parts = [first]
for j in range(doc_start + 1, len(lines)):
parts.append(lines[j])
if quote in lines[j]:
break
result[idx] = "\n".join(parts)
return result
class QualityChecker:
"""Main quality checker that runs all checks."""
def __init__(self):
self.checkers = [
EmojiChecker(),
LanguageChecker(),
DocstringChecker(),
ImportChecker(),
CommentChecker(),
GetAttrChecker(),
]
self.gitignore_spec = self._load_gitignore()
def _load_gitignore(self):
"""Load .gitignore patterns for filtering files."""
if not pathspec:
return None
gitignore_path = Path(".gitignore")
if not gitignore_path.exists():
return None
try:
with open(gitignore_path, encoding="utf-8") as f:
return pathspec.PathSpec.from_lines("gitwildmatch", f)
except Exception:
return None
def _should_ignore_file(self, file_path: str) -> bool:
"""Check if file should be ignored based on gitignore."""
if not self.gitignore_spec:
return False
# Convert to relative path for gitignore matching
try:
rel_path = os.path.relpath(file_path)
return self.gitignore_spec.match_file(rel_path)
except Exception:
return False
def check_files(self, file_paths: list[str]) -> list[Violation]:
"""Run all checks on the given files."""
all_violations = []
# Filter files that exist and have relevant extensions
valid_files = []
for file_path in file_paths:
# Skip this script to avoid self-checking issues
if file_path.endswith("quality_check.py"):
continue
# Skip files ignored by gitignore
if self._should_ignore_file(file_path):
continue
if os.path.isfile(file_path):
ext = os.path.splitext(file_path)[1].lower()
if ext in ALL_EXTENSIONS:
valid_files.append(file_path)
if not valid_files:
return all_violations
# Process files in parallel for better performance
def check_single_file(file_path):
file_violations = []
for checker in self.checkers:
file_violations.extend(checker.check_file(file_path))
return file_violations
# Use ThreadPoolExecutor for parallel processing
with ThreadPoolExecutor(max_workers=8) as executor: # Increased workers
# Submit all file checking tasks
future_to_file = {
executor.submit(check_single_file, file_path): file_path
for file_path in valid_files
}
completed = 0
for future in as_completed(future_to_file):
completed += 1
if completed % 10 == 0 or completed == len(valid_files):
logger.info(f"Progress: {completed}/{len(valid_files)} files checked")
try:
file_violations = future.result()
all_violations.extend(file_violations)
except Exception as e:
file_path = future_to_file[future]
logger.error(f"Error checking {file_path}: {e}")
return all_violations
def get_modified_files(self) -> list[str]:
"""Get list of modified files from git."""
import os
import subprocess
try:
# In CI/PR context, compare against target branch
if os.getenv("GITHUB_EVENT_NAME") == "pull_request":
base_ref = os.getenv("GITHUB_BASE_REF", "main")
result = subprocess.run(
["git", "diff", "--name-only", f"origin/{base_ref}...HEAD"],
capture_output=True,
text=True,
check=True,
)
modified_files = result.stdout.strip().split("\n") if result.stdout.strip() else []
return [f for f in modified_files if f]
# Local development: check staged, unstaged, and untracked files
# Get staged files
result = subprocess.run(
["git", "diff", "--cached", "--name-only"],
capture_output=True,
text=True,
check=True,
)
staged_files = result.stdout.strip().split("\n") if result.stdout.strip() else []
# Get unstaged files
result = subprocess.run(
["git", "diff", "--name-only"],
capture_output=True,
text=True,
check=True,
)
unstaged_files = result.stdout.strip().split("\n") if result.stdout.strip() else []
# Get untracked files
result = subprocess.run(
["git", "ls-files", "--others", "--exclude-standard"],
capture_output=True,
text=True,
check=True,
)
untracked_files = result.stdout.strip().split("\n") if result.stdout.strip() else []
# Combine all files
all_files = list(set(staged_files + unstaged_files + untracked_files))
return [f for f in all_files if f] # Filter out empty strings
except subprocess.SubprocessError:
logger.warning("Failed to get modified files from git. Checking all files.")
return []
def main():
"""Run comprehensive code quality checks with configurable options."""
"""Run comprehensive code quality checks with configurable options."""
"""Main entry point for the quality checker."""
parser = argparse.ArgumentParser(description="Professional Quality Check Tool")
parser.add_argument("--fix", action="store_true", help="Attempt to automatically fix issues")
parser.add_argument(
"--strict", action="store_true", help="Exit with error code on any violation"
)
parser.add_argument("--files", nargs="+", help="Specific files to check")
parser.add_argument("--all", action="store_true", help="Check all files in repository")
args = parser.parse_args()
checker = QualityChecker()
# Determine which files to check
if args.files:
files_to_check = args.files
elif args.all:
# Check all relevant files in repository (deterministic)
files_to_check = []
from pathlib import Path
import pathspec
# Load .gitignore patterns
gitignore_path = Path(".gitignore")
if gitignore_path.exists():
with open(gitignore_path, encoding="utf-8") as f:
spec = pathspec.PathSpec.from_lines("gitwildmatch", f)
else:
spec = pathspec.PathSpec.from_lines("gitwildmatch", [])
for pattern in [
"**/*.py",
"**/*.md",
"**/*.rst",
"**/*.txt",
"**/*.yaml",
"**/*.yml",
"**/*.json",
"**/*.toml",
]:
for file_path in Path(".").rglob(pattern):
if file_path.is_file():
# Check if file should be ignored
rel_path = file_path.relative_to(Path("."))
if not spec.match_file(str(rel_path)):
files_to_check.append(str(file_path))
files_to_check = sorted(files_to_check)
else:
# Check only git modified files
files_to_check = checker.get_modified_files()
# Run checks
violations = checker.check_files(files_to_check)
# Print results
if violations:
logger.error(f"\n{len(violations)} quality issues found:\n")
# Group by file and count by category
violations_by_file = {}
category_counts = {}
for v in violations:
if v.file_path not in violations_by_file:
violations_by_file[v.file_path] = []
violations_by_file[v.file_path].append(v)
# Count by category (extract category from message)
if "Hyperbolic term" in v.message:
category = "Hyperbolic terms"
elif "Debug print/logging statement" in v.message:
category = "Debug print statements"
elif "Unused imports" in v.message:
category = "Unused imports"
elif "Commented-out code" in v.message:
category = "Commented-out code"
elif "getattr should only" in v.message:
category = "Unjustified getattr"
else:
category = "Other issues"
category_counts[category] = category_counts.get(category, 0) + 1
# Print violations by file
for file_path, file_violations in violations_by_file.items():
logger.error(f"\n{file_path}: ({len(file_violations)} issues)")
for v in sorted(file_violations, key=lambda v: v.line_num):
logger.error(f" Line {v.line_num}: {v.message}")
logger.error(f" {v.content}")
# Print summary by category
logger.error("\n" + "-" * 40)
logger.error("Summary:")
for category, count in sorted(category_counts.items()):
logger.error(f"{category}: {count}")
logger.error("-" * 40)
logger.error(f"Total files with issues: {len(violations_by_file)}")
logger.error(f"Total issues: {len(violations)}")
# Exit with error if strict mode
if args.strict:
sys.exit(1)
else:
logger.info("No quality issues found!")
sys.exit(0)
if __name__ == "__main__":
main()