AutoRpa/rpa_contract/contract_number.py

607 lines
23 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# 合同 S13:按 fdId 打开详情读正式合同号。
#
# fdId 优先于标题+客户编码。禁止盲取「我的合同」最新一条。
from __future__ import annotations
import time
from contract_common import (
BrowserAdapter,
ContractFlowContext,
_expected_title,
_is_logged_in,
_open_submitted_view,
_read_document_id,
_usable_contract_no,
)
from contract_login import login_contract_system
# 读正式合同号(S13)。没有 fdId 则返回空字符串。
def read_contract_number(
browser: BrowserAdapter, ctx: ContractFlowContext | None = None, *, fd_id: str = ""
) -> str:
no, _via = _read_contract_number_routed(browser, ctx, fd_id=fd_id)
return no
# 按 fdId 打开详情读号。读不到返回空,不回列表盲取最新一条。
def _read_contract_number_routed(
browser: BrowserAdapter, ctx: ContractFlowContext | None = None, *, fd_id: str = ""
) -> tuple[str, str]:
del ctx
doc_id = (fd_id or "").strip() or _read_document_id(browser)
if not doc_id:
print("S13 没有 fdId,无法打开详情", flush=True)
return "", "none"
print(f"S13:开详情读号 fdId={doc_id}", flush=True)
no = _read_number_by_fd_id(browser, doc_id)
if no:
return no, "docNumber"
print("S13 详情未读到合同编号", flush=True)
return "", "none"
# 从详情页 #_xform_docNumber 读合同编号。占位符和手机号不算。
def _read_contract_number_on_page(browser: BrowserAdapter) -> str:
try:
browser.switch_to_default_content()
except Exception:
pass
scanned = browser.execute_script(
"""
const skip = ['提交时自动生成', '自动生成', '==请选择==', '合同编号', '合同标题'];
function usable(raw) {
const v = String(raw || '').replace(/\\s+/g, ' ').trim();
if (!v || skip.indexOf(v) >= 0) return '';
return v;
}
function fromDoc(doc) {
const node = doc.querySelector('#_xform_docNumber');
if (node) {
const typed = node.querySelector('input, textarea');
const v = usable(typed ? typed.value : (node.innerText || node.textContent || ''));
if (v) return v;
}
const input = doc.querySelector('input[name="docNumber"]');
if (input) {
const v = usable(input.value || '');
if (v) return v;
}
return '';
}
function walk(win) {
try {
const no = fromDoc(win.document);
if (no) return no;
} catch (e) {}
for (let i = 0; i < win.frames.length; i++) {
try {
const no = walk(win.frames[i]);
if (no) return no;
} catch (e) {}
}
return '';
}
return walk(window);
"""
)
if isinstance(scanned, str) and _usable_contract_no(scanned.strip()):
return scanned.strip()
if scanned:
print(f"S13 详情页扫到不可用值={scanned!r}", flush=True)
return ""
# 把当前页导航回合同系统主界面,等到出现「我的合同」。
def _reload_contract_home(browser: BrowserAdapter, home: str) -> bool:
try:
browser.switch_to_default_content()
except Exception:
pass
print("刷新回合同主界面:", home, flush=True)
try:
browser.execute_script(
"try { window.top.location.href = arguments[0]; } catch (e) { location.href = arguments[0]; } return true;",
home,
)
except Exception:
pass
try:
browser.open_url(home)
except Exception as exc:
print(f"open_url 主界面失败:{exc}", flush=True)
return _wait_text_in_frames(browser, "我的合同", timeout_seconds=12)
# 刷新主界面失败时关页重开;掉登录则再登一次。
def _reopen_contract_home(browser: BrowserAdapter, ctx: ContractFlowContext, home: str) -> bool:
print("主界面刷新失败,关闭当前页后重开", flush=True)
reopen = getattr(browser, "reopen_url", None)
if callable(reopen):
reopen(home)
else:
browser.open_url(home)
if _wait_text_in_frames(browser, "我的合同", timeout_seconds=12):
return True
if not _is_logged_in(browser):
login_contract_system(browser, ctx)
browser.open_url(home)
return _wait_text_in_frames(browser, "我的合同", timeout_seconds=12)
# 回到主界面并点开「我的合同 / 我起草的」。
def _open_my_contracts(browser: BrowserAdapter, ctx: ContractFlowContext) -> None:
home = (ctx.contract_url or "").strip() or "http://contract-test.gzmpc.com:8080/"
if not _reload_contract_home(browser, home):
_reopen_contract_home(browser, ctx, home)
if not _click_my_contracts_sidebar(browser):
print("未点到侧栏「我的合同」,关闭页面后重开", flush=True)
_reopen_contract_home(browser, ctx, home)
_click_my_contracts_sidebar(browser)
_wait_text_in_frames(browser, "我起草的", timeout_seconds=10)
_click_by_exact_text(browser, "我起草的")
_wait_text_in_frames(browser, "申请时间", timeout_seconds=10)
_wait_text_in_frames(browser, "合同编号", timeout_seconds=10)
# 点侧栏「我的合同」。点到后等到「我起草的」。
def _click_my_contracts_sidebar(browser: BrowserAdapter) -> bool:
clicked = browser.execute_script(
"""
const label = '我的合同';
function clickIn(win) {
const nodes = Array.from(win.document.querySelectorAll('a, span, div, li, p, em, td'));
const hits = nodes.filter((el) => (el.textContent || '').trim() === label);
hits.sort((a, b) => {
const ax = a.getBoundingClientRect().left;
const bx = b.getBoundingClientRect().left;
if (ax !== bx) return ax - bx;
return a.children.length - b.children.length;
});
if (hits.length) { hits[0].click(); return true; }
for (let i = 0; i < win.frames.length; i++) {
try { if (clickIn(win.frames[i])) return true; } catch (e) {}
}
return false;
}
return clickIn(window);
"""
)
if clicked is True:
return _wait_text_in_frames(browser, "我起草的", timeout_seconds=8)
_click_by_exact_text(browser, "我的合同")
return _wait_text_in_frames(browser, "我起草的", timeout_seconds=8)
# 在当前页和 iframe 里等到出现指定文本。超时返回 False。
def _wait_text_in_frames(browser: BrowserAdapter, text: str, timeout_seconds: int = 15) -> bool:
deadline = time.time() + timeout_seconds
while time.time() < deadline:
if _page_has_text(browser, text):
return True
time.sleep(0.3)
return False
# 当前页或任一 iframe 正文是否含有指定文本。
def _page_has_text(browser: BrowserAdapter, text: str) -> bool:
found = browser.execute_script(
"""
const wanted = String(arguments[0] || '');
function hasTextInFrames(win) {
try {
const body = (win.document && win.document.body && win.document.body.innerText) || '';
if (body.indexOf(wanted) >= 0) return true;
} catch (e) {}
for (let i = 0; i < win.frames.length; i++) {
try { if (hasTextInFrames(win.frames[i])) return true; } catch (e) {}
}
return false;
}
return hasTextInFrames(window);
""",
text,
)
return found is True
# 按可见文本精确点击。叶子节点优先。
def _click_by_exact_text(browser: BrowserAdapter, label: str) -> None:
clicked = browser.execute_script(
"""
const label = String(arguments[0] || '');
function clickIn(doc) {
const nodes = Array.from(doc.querySelectorAll('div, a, span, li, p, em, td, th, button'));
const leaf = nodes.find((el) => (el.textContent || '').trim() === label && el.children.length === 0);
if (leaf) { leaf.click(); return true; }
const parent = nodes.find((el) => (el.textContent || '').trim() === label);
if (parent) { parent.click(); return true; }
return false;
}
function clickInFrames(win) {
try { if (clickIn(win.document)) return true; } catch (e) {}
for (let i = 0; i < win.frames.length; i++) {
try { if (clickInFrames(win.frames[i])) return true; } catch (e) {}
}
return false;
}
return clickInFrames(window) ? label : '';
""",
label,
)
if clicked != label:
try:
browser.click_first([f"text={label}"])
except Exception:
pass
# 按申请时间排序「我的合同」列表。
def _sort_my_contracts_by_apply_time(browser: BrowserAdapter) -> None:
_click_by_exact_text(browser, "申请时间")
browser.execute_script(
"""
const latestByApplyTime = true;
function textOf(el) { return (el.textContent || '').replace(/\\s+/g, ' ').trim(); }
function sortIn(doc) {
const desc = Array.from(doc.querySelectorAll('[title], a, span, i, em')).find((el) => {
const t = (el.getAttribute('title') || el.textContent || '').trim();
return t === '降序' || t === '从晚到早' || t === '倒序';
});
if (desc) { desc.click(); return 'desc'; }
const bar = Array.from(doc.querySelectorAll('div, span')).find((el) => {
const t = textOf(el);
return t.indexOf('更多排序') >= 0 && t.length < 80;
});
if (bar) {
const downs = Array.from(bar.querySelectorAll('i, em, span, a')).filter((el) => {
const t = (el.textContent || '').trim();
const cls = String(el.className || '');
return t === '∨' || t === 'v' || t === '↓' || t === '▼'
|| cls.indexOf('down') >= 0 || cls.indexOf('desc') >= 0;
});
if (downs.length) { downs[downs.length - 1].click(); return 'arrow'; }
}
return '';
}
function walk(win) {
try {
const hit = sortIn(win.document);
if (hit) return hit;
} catch (e) {}
for (let i = 0; i < win.frames.length; i++) {
try {
const hit = walk(win.frames[i]);
if (hit) return hit;
} catch (e) {}
}
return '';
}
return walk(window) || (latestByApplyTime ? 'apply-time' : '');
"""
)
_wait_text_in_frames(browser, "合同编号", timeout_seconds=10)
# 刮「我的合同」第一行可用合同号。给排查用,正式读号不走这里。
def _scrape_latest_my_contract_number(browser: BrowserAdapter) -> str:
scanned = browser.execute_script(
"""
const latestByApplyTime = true;
const skip = ['提交时自动生成', '自动生成', '==请选择=='];
function scrape(doc) {
const rows = Array.from(doc.querySelectorAll('tr'));
let headerIdx = null;
for (const tr of rows) {
const cells = Array.from(tr.querySelectorAll('th, td')).map((c) =>
(c.innerText || '').replace(/\\s+/g, ' ').trim()
);
if (!headerIdx && cells.indexOf('合同编号') >= 0) {
headerIdx = {};
cells.forEach((h, i) => { if (h && headerIdx[h] === undefined) headerIdx[h] = i; });
continue;
}
if (!headerIdx) continue;
const no = (cells[headerIdx['合同编号']] || '').trim();
if (!no || skip.indexOf(no) >= 0) continue;
return latestByApplyTime ? no : no;
}
return '';
}
function walk(win) {
try {
const no = scrape(win.document);
if (no) return no;
} catch (e) {}
for (let i = 0; i < win.frames.length; i++) {
try {
const no = walk(win.frames[i]);
if (no) return no;
} catch (e) {}
}
return '';
}
return walk(window);
"""
)
if isinstance(scanned, str) and _usable_contract_no(scanned.strip()):
return scanned.strip()
return ""
# 从列表行里按 fdId、标题+客户编码挑合同号。fdId 优先。
def _pick_contract_row(
rows: list[dict],
*,
fd_id: str = "",
title: str = "",
customer_code: str = "",
) -> str:
fd_id = (fd_id or "").strip()
title = (title or "").strip()
customer_code = (customer_code or "").strip()
if fd_id:
for row in rows:
href = str(row.get("href") or "")
no = str(row.get("no") or "").strip()
if fd_id in href and _usable_contract_no(no):
return no
if title and customer_code:
for row in rows:
row_title = str(row.get("title") or "")
row_cust = str(row.get("customer") or "")
no = str(row.get("no") or "").strip()
if title in row_title and customer_code in row_cust and _usable_contract_no(no):
return no
if title:
for row in rows:
no = str(row.get("no") or "").strip()
if title in str(row.get("title") or "") and _usable_contract_no(no):
return no
return ""
# 在「我的合同」查询里填标题和客户编码并点搜索。
def _search_my_contracts(browser: BrowserAdapter, title: str, customer_code: str) -> None:
title = (title or "").strip()
customer_code = (customer_code or "").strip()
if not title and not customer_code:
return
browser.execute_script(
"""
const title = String(arguments[0] || '');
const customer = String(arguments[1] || '');
function setValue(el, value) {
if (!el) return false;
el.focus();
el.value = value;
el.dispatchEvent(new Event('input', {bubbles: true}));
el.dispatchEvent(new Event('change', {bubbles: true}));
return true;
}
function fillNearLabel(doc, label, value) {
if (!value) return false;
const nodes = Array.from(doc.querySelectorAll('td, th, span, label, div, em'));
for (const node of nodes) {
const text = (node.innerText || '').replace(/\\s+/g, '').trim();
if (text !== label && text !== label + ':' && text !== label + ':') continue;
const box = node.parentElement || node;
const input = box.querySelector('input[type="text"], input:not([type]), input[type="search"]');
if (setValue(input, value)) return true;
const next = node.nextElementSibling;
if (next && setValue(next.querySelector('input'), value)) return true;
}
return false;
}
function fillPlaceholder(doc, hint, value) {
if (!value) return false;
const inputs = Array.from(doc.querySelectorAll('input'));
const el = inputs.find((i) => (i.getAttribute('placeholder') || '').indexOf(hint) >= 0);
if (!el) return false;
setValue(el, value);
const btn = el.parentElement && el.parentElement.querySelector('a, button, span, i, em');
if (btn) btn.click();
return true;
}
function clickSearch(doc) {
const labels = ['搜索', '查询'];
const nodes = Array.from(doc.querySelectorAll('a, div, span, button, input'));
for (const label of labels) {
const hit = nodes.find((el) => {
const text = (el.textContent || el.value || '').trim();
return text === label && (!el.children || el.children.length === 0);
});
if (hit) { hit.click(); return true; }
}
return false;
}
function walk(win) {
try {
fillPlaceholder(win.document, '合同(协议)全称', title)
|| fillPlaceholder(win.document, '合同标题', title)
|| fillNearLabel(win.document, '合同标题', title)
|| fillNearLabel(win.document, '合同(协议)全称', title);
fillPlaceholder(win.document, '客户编码', customer)
|| fillNearLabel(win.document, '客户编码', customer);
clickSearch(win.document);
} catch (e) {}
for (let i = 0; i < win.frames.length; i++) {
try { walk(win.frames[i]); } catch (e) {}
}
}
walk(window);
return true;
""",
title,
customer_code,
)
_wait_text_in_frames(browser, "合同编号", timeout_seconds=10)
# 刮「我的合同」表格:编号、标题、客户、带 fdId 的链接。
def _scrape_my_contract_rows(browser: BrowserAdapter) -> list[dict]:
raw = browser.execute_script(
"""
function scrapeTable(doc) {
const rows = Array.from(doc.querySelectorAll('tr'));
let header = null;
const found = [];
for (const tr of rows) {
const cells = Array.from(tr.querySelectorAll('th, td')).map((c) =>
(c.innerText || '').replace(/\\s+/g, ' ').trim()
);
if (!header && cells.indexOf('合同编号') >= 0) {
header = {};
cells.forEach((h, i) => { if (h && header[h] === undefined) header[h] = i; });
continue;
}
if (!header) continue;
const no = (cells[header['合同编号']] || '').trim();
const title = (
cells[header['合同标题']]
|| cells[header['合同(协议)全称']]
|| cells[header['合同名称']]
|| cells[header['主题']]
|| cells[header['标题']]
|| ''
).trim();
const customer = (
cells[header['客户编码']] || cells[header['客户名称']] || cells[header['客商编码']] || ''
).trim();
const link = tr.querySelector('a[href*="fdId"], a[href*="fdid"]');
const box = tr.querySelector('input[name="List_Selected"]');
const href = link
? String(link.getAttribute('href') || '')
: (box ? String(box.value || '') : (tr.innerHTML || ''));
if (!no && !href) continue;
found.push({no: no, title: title, customer: customer, href: href});
}
return found;
}
function scrapeLinks(doc) {
const found = [];
const links = Array.from(doc.querySelectorAll('a[href*="fdId"], a[href*="fdid"]'));
for (const a of links) {
const href = String(a.getAttribute('href') || '');
const tr = a.closest('tr') || a.parentElement;
const blob = ((tr && tr.innerText) || a.innerText || '').replace(/\\s+/g, ' ').trim();
const hit = blob.match(/HT[0-9A-Za-z-]+/) || blob.match(/\\b\\d{10,}\\b/);
found.push({
no: hit ? hit[0] : '',
title: blob,
customer: blob,
href: href
});
}
return found;
}
function walk(win) {
let all = [];
try { all = all.concat(scrapeTable(win.document)); } catch (e) {}
try { all = all.concat(scrapeLinks(win.document)); } catch (e) {}
for (let i = 0; i < win.frames.length; i++) {
try { all = all.concat(walk(win.frames[i])); } catch (e) {}
}
return all;
}
return walk(window);
"""
)
if not isinstance(raw, list):
return []
rows: list[dict] = []
for item in raw:
if not isinstance(item, dict):
continue
rows.append(
{
"no": str(item.get("no") or "").strip(),
"title": str(item.get("title") or "").strip(),
"customer": str(item.get("customer") or "").strip(),
"href": str(item.get("href") or "").strip(),
}
)
return rows
# 列表行的 href 里是否含有该 fdId。
def _row_has_fd_id(rows: list[dict], fd_id: str) -> bool:
token = (fd_id or "").strip()
if not token:
return False
return any(token in str(row.get("href") or "") for row in rows)
# 打开该 fdId 的详情页并读合同编号。文档不存在则返回空。
def _read_number_by_fd_id(browser: BrowserAdapter, fd_id: str) -> str:
doc_id = (fd_id or "").strip()
if not doc_id:
return ""
_open_submitted_view(browser, doc_id)
deadline = time.time() + 10
while time.time() < deadline:
body = browser.read_body_text() or ""
if "该文档不存在" in body or "操作失败" in body:
print(f"S13 详情不存在 fdId={doc_id}", flush=True)
return ""
no = _read_contract_number_on_page(browser)
if no:
print("从合同详情读到合同编号:", no, flush=True)
return no
time.sleep(0.8)
return ""
# 回「我的合同」按 fdId/标题+客户编码查号。匹配不到不取最新一条。
def _read_contract_number_from_my_list(
browser: BrowserAdapter,
ctx: ContractFlowContext,
timeout_seconds: int = 25,
fd_id: str = "",
) -> str:
title = _expected_title(ctx)
customer = (ctx.customer_code or "").strip()
print(
f"S13:回我的合同查询 fdId={fd_id or '-'} title={title} customer={customer}",
flush=True,
)
_open_my_contracts(browser, ctx)
deadline = time.time() + timeout_seconds
searched = False
saw_row = False
while time.time() < deadline:
rows = _scrape_my_contract_rows(browser)
print(f"S13 列表行数={len(rows)}", flush=True)
for row in rows[:6]:
print(
"S13 row"
f" no={row.get('no')!r}"
f" title={str(row.get('title') or '')[:60]!r}"
f" customer={str(row.get('customer') or '')[:30]!r}"
f" href={str(row.get('href') or '')[:90]!r}",
flush=True,
)
no = _pick_contract_row(
rows,
fd_id=fd_id,
title=title,
customer_code=customer,
)
if no:
print("从我的合同读到合同编号:", no, flush=True)
return no
if fd_id and _row_has_fd_id(rows, fd_id):
saw_row = True
if not searched:
_search_my_contracts(browser, title, customer)
searched = True
time.sleep(1)
if saw_row and fd_id:
no = _read_number_by_fd_id(browser, fd_id)
if no:
return no
print("我的合同列表未匹配到合同编号", flush=True)
return ""