Merge one Scholar page parse into the accumulated profile snapshot.
(base: Dict[str, Any], incoming: Dict[str, Any], publication_limit: int)
| 1202 | |
| 1203 | |
| 1204 | def _merge_scholar_page_data(base: Dict[str, Any], incoming: Dict[str, Any], publication_limit: int) -> Dict[str, Any]: |
| 1205 | """Merge one Scholar page parse into the accumulated profile snapshot.""" |
| 1206 | if not base: |
| 1207 | base = { |
| 1208 | "name": incoming.get("name", ""), |
| 1209 | "affiliation": incoming.get("affiliation", ""), |
| 1210 | "homepage_url": incoming.get("homepage_url", ""), |
| 1211 | "interests": list(incoming.get("interests", []) or []), |
| 1212 | "publications": [], |
| 1213 | "all_publications": [], |
| 1214 | "stats": dict(incoming.get("stats") or {}), |
| 1215 | "top_coauthors": [], |
| 1216 | "top_venues": [], |
| 1217 | "collaboration_network": [], |
| 1218 | "top_cited_publications": [], |
| 1219 | "blocked": bool(incoming.get("blocked")), |
| 1220 | } |
| 1221 | else: |
| 1222 | if not base.get("name") and incoming.get("name"): |
| 1223 | base["name"] = incoming["name"] |
| 1224 | if not base.get("affiliation") and incoming.get("affiliation"): |
| 1225 | base["affiliation"] = incoming["affiliation"] |
| 1226 | if not base.get("homepage_url") and incoming.get("homepage_url"): |
| 1227 | base["homepage_url"] = incoming["homepage_url"] |
| 1228 | if not base.get("stats") and incoming.get("stats"): |
| 1229 | base["stats"] = dict(incoming.get("stats") or {}) |
| 1230 | base["blocked"] = bool(base.get("blocked")) or bool(incoming.get("blocked")) |
| 1231 | for interest in incoming.get("interests", []) or []: |
| 1232 | if interest and interest not in base["interests"]: |
| 1233 | base["interests"].append(interest) |
| 1234 | |
| 1235 | seen_titles = {str(item.get("title") or "").casefold() for item in base.get("all_publications", [])} |
| 1236 | for publication in incoming.get("all_publications") or incoming.get("publications") or []: |
| 1237 | title_key = str(publication.get("title") or "").casefold() |
| 1238 | if not title_key or title_key in seen_titles: |
| 1239 | continue |
| 1240 | seen_titles.add(title_key) |
| 1241 | base.setdefault("all_publications", []).append(publication) |
| 1242 | |
| 1243 | sorted_publications = sorted( |
| 1244 | base.get("all_publications", []), |
| 1245 | key=lambda item: ( |
| 1246 | int(item.get("citations") or 0), |
| 1247 | int(item.get("year") or 0), |
| 1248 | len(str(item.get("title") or "")), |
| 1249 | ), |
| 1250 | reverse=True, |
| 1251 | ) |
| 1252 | base["all_publications"] = sorted_publications |
| 1253 | base["publications"] = sorted_publications[:publication_limit] |
| 1254 | |
| 1255 | network_signals = _build_scholar_network_signals(sorted_publications, scholar_name=str(base.get("name") or "")) |
| 1256 | base["top_coauthors"] = list(network_signals.get("top_coauthors") or []) |
| 1257 | base["top_venues"] = list(network_signals.get("top_venues") or []) |
| 1258 | base["collaboration_network"] = list(network_signals.get("collaboration_network") or []) |
| 1259 | base["top_cited_publications"] = list(network_signals.get("top_cited_publications") or []) |
| 1260 | return base |
| 1261 |
no test coverage detected