This repository was archived by the owner on Dec 30, 2025. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathoperations.py
More file actions
159 lines (136 loc) · 6.13 KB
/
Copy pathoperations.py
File metadata and controls
159 lines (136 loc) · 6.13 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
from fastapi import APIRouter, HTTPException, Path
from utility import get_translation
from starlette import status
import table, infoBox, header, citation, images
from pydantic import BaseModel, Field
operations_router = APIRouter(
prefix="/operations",
tags=["Analysis"],
)
LANGUAGES = {
"en": "English",
"es": "Spanish (Español)",
"fr": "French (Français)",
"de": "German (Deutsch)",
"pt": "Portuguese (Português)",
"ar": "Arabic (العربية)"
}
class FinalResponse(BaseModel):
title:str = Field(title="Final Page Title")
table_analysis:table.TableResponse = Field(title="Table Analysis")
header_analysis:header.HeaderCount = Field(title="Header Analysis")
info_box:infobox.InfoBoxResponse = Field(title="InfoBox Analysis")
citations:citation.CitationResponse = Field(title="Citation Analysis")
total_images:int = Field(title="Total images/Media Files")
def calculate_single_score(article_response: FinalResponse) -> float:
"""Calculates the combined quality score for a single article's response object."""
total_tables = article_response.table_analysis.number_of_tables
total_infobox_attrs = article_response.info_box.total_attributes
total_citations = article_response.citations.total_citations
total_headers = article_response.header_analysis.total_count
total_images = article_response.total_images
# Structural scoring formula:
score = ((0.5 * total_citations) + (0.3 * total_tables) +
(0.10 * total_infobox_attrs) + (0.05 * total_headers) + (0.05 * total_images))
return score
# --- Helper Function 2: Single Article Analysis ---
def analyze_single_article(title: str, language: str) -> FinalResponse:
"""Performs all structural analyses (table, header, infobox, citation) for a single article."""
try:
# Perform all analysis calls
table_analysis = table.analyze_tables(title, language)
header_counter = header.count_html_headers(title, language)
infobox_analysis = infobox.analyze_infobox(title, language)
citation_analysis = citation.extract_citation_from_wikitext(title, language)
image_count = images.get_image_count(title, language)
return FinalResponse(
title=title,
table_analysis=table_analysis,
header_analysis=header_counter,
info_box=infobox_analysis,
citations=citation_analysis,
total_images=image_count
)
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Structural analysis error for {title} ({language}): {str(e)}")
# --- Main API Endpoint (Modified) ---
# REMOVED the redundant '{language}' path parameter
@operations_router.get("/{source_language}/{title}", status_code=status.HTTP_200_OK)
async def get_results(title: str, source_language: str = Path(min_length=1)):
"""
Analyzes the structural quality score for the given article across all 6 supported languages.
"""
all_scores = []
normalized_title = title.replace(" ", "_")
target_languages = list(LANGUAGES.keys())
for lang_code in target_languages:
current_title = ""
# 1. Determine Title (Source vs. Translation)
if lang_code == source_language:
# For the user's source language, use the original title
current_title = normalized_title
else:
# For all other languages, attempt translation
current_title = get_translation(normalized_title, source_language, lang_code)
# 2. Check for Translation Success
if not current_title:
all_scores.append({
"lang_code": lang_code,
"lang_name": LANGUAGES[lang_code],
"title": None,
"score": -1, # -1 indicates the article could not be found/translated
"is_user_language": lang_code == source_language,
"is_authority_article": False,
"error": "Translation or article not available."
})
continue
try:
# 3. Analyze and Score
article_response = analyze_single_article(current_title, lang_code)
score = calculate_single_score(article_response)
# 4. Store Result
all_scores.append({
"lang_code": lang_code,
"lang_name": LANGUAGES[lang_code],
"title": current_title,
"score": round(score, 3),
"is_user_language": lang_code == source_language,
"is_authority_article": False
})
except HTTPException as e:
# Handle analysis errors (e.g., 404 from a downstream function)
all_scores.append({
"lang_code": lang_code,
"lang_name": LANGUAGES[lang_code],
"title": current_title,
"score": -1,
"is_user_language": lang_code == source_language,
"is_authority_article": False,
"error": e.detail
})
except Exception as e:
# Handle unexpected errors
all_scores.append({
"lang_code": lang_code,
"lang_name": LANGUAGES[lang_code],
"title": current_title,
"score": -1,
"is_user_language": lang_code == source_language,
"is_authority_article": False,
"error": f"Internal Error during analysis: {str(e)}"
})
valid_scores = [d['score'] for d in all_scores if d.get('score', -1) >= 0]
max_score = max(valid_scores) if valid_scores else -float('inf')
for item in all_scores:
is_authority = (item.get('score', -1) >= 0) and (item.get('score') == max_score)
item['is_authority_article'] = is_authority
sorted_scores = sorted(all_scores, key=lambda x: x.get('score', -float('inf')), reverse=True)
# 5. Return the combined results
return {
"article": title,
"source_language_code": source_language,
"scores_by_language": sorted_scores
}