Unify configuration-driven DMS connection modes

docs/dms-schema.md

@@ -8,8 +8,20 @@
 
 程序仍先按栏目 ID 读取元数据，核对栏目名、父栏目和模型字段类型，再访问内容。字段名映射应用于查询条件、写入内容和读取结果；配置非法或服务端模型不符会停止相应 DMS 操作。`sh deploy-offline.sh check-dms`也读取本文件及现有环境变量，按实际映射只读核对五个栏目的元数据。修改配置不能替代服务端栏目、模型和历史数据迁移。离线部署须重新生成单包，旧包不会自动包含新配置。
 
-新环境可在`.env`中设置`DMS_BASE_URL`为服务主机、`AUTH_PORT`为认证端口、`DMS_PORT`为数据端口。客户端使用账号密码向认证端口的`/proxy_oauth/user/login`获取token，token仅保存在内存，再向数据端口的`/proxy_dms`访问栏目与模型；未设置端口时沿用`DMS_BASE_URL`的端口。HTTP 401或DMS 212时有界重新登录；无需手动获取token。直接`/dms`且未配置认证端口时仍要求提供token，避免猜测认证地址。`check-dms`只验证栏目元数据和字段类型，不因栏目暂无记录而失败。
+本地和线上共用同一套客户端与`.env`配置项，不根据Git分支、主机名或字段前缀选择代码版本。`connection_urls`统一解析初次登录、业务请求和token刷新使用的地址；企业查询与会话分析共用这一入口。
 
-2026-09-30的`dev`本地配置已依据当前服务核验：父栏目“营商助手”1883，助手问答记录1889、助手会话1887、诉求词1891、企业基础信息1888、企业荣誉信息1886。五个名称均与真实元数据一致；65个字段映射value各删除一层`c_`并通过模型类型核验。其他环境应重新核验，不能直接复用这些ID。
+| 配置方式 | 登录请求 | 数据请求 |
+| --- | --- | --- |
+| `DMS_BASE_URL`为主机和可选端口，`AUTH_PORT`/`DMS_PORT`留空 | 原端口的`/proxy_oauth/user/login` | 原端口的`/proxy_dms` |
+| 设置`AUTH_PORT`与`DMS_PORT` | 认证端口的`/proxy_oauth/user/login` | 数据端口的`/proxy_dms` |
+| 只设置其中一个端口 | 对应端口覆盖，另一项沿用URL端口 | 对应端口覆盖，另一项沿用URL端口 |
+| URL显式含`/proxy_dms` | 沿用同样的端口规则 | 保留`/proxy_dms`路径 |
+| URL显式含`/dms` | 有`AUTH_PORT`可自动登录；未设置时使用已有`DMS_TOKEN`，不猜认证地址 | 保留`/dms`路径，端口规则相同 |
+
+`DMS_USERNAME`和`DMS_PASSWORD`用于自动登录；已提供`DMS_TOKEN`则先使用它。HTTP 401或DMS 212时，仅在已配置可用登录方式的情况下重新登录并重试一次；无权限或不确定的传输错误不重放写请求。token只留内存，日志不输出凭据。配置项名称相同，值按环境填写；不需要为本地和线上分别维护客户端源码。
+
+栏目ID、`c_name`/`c_c_name`等实际字段差异始终由各环境`config/dms_schema.json`映射，程序不自动增删前缀。仓库默认映射沿用dev版本；线上已有独立config与.env时继续使用现场文件，更新代码不代表迁移栏目或替换这些配置。`check-dms`按当前配置验证栏目元数据和字段类型，不因栏目暂无记录而失败。
 
 会话与企业列表共用精确空结果契约：仅`selectContentList`返回`code=202`且`message`与`content`均为“数据不存在”时视为空列表。其他202、权限/认证失败、元数据和写入错误仍明确失败。真实合成草稿已验证新增、更新及回读；销毁请求`state=4`获成功响应，草稿列表确认测试记录退出。当前服务未返回销毁状态的记录，因此尚未验证销毁记录的归档回读或可恢复性；不能把销毁操作称为物理删除。
+
+F131验证：同一组配置项下，六种地址/端口组合、两种字段映射、企业与会话客户端交叉验证，覆盖200标准空列表、精确202空列表、认证刷新和权限/未知错误隔离。全部使用合成服务替身；不代表已在两套真实环境完成联调，也不修改线上独立配置。

src/step5_session_analysis/dms.py

@@ -123,18 +123,30 @@ class NoRedirect(HTTPRedirectHandler):
         raise DmsError('dms_redirect_refused')
 
 
+def connection_urls(config):
+    """One configuration contract for local, split-port and direct deployments.
+
+    Explicit ports override the base port independently. Direct /dms has no
+    inferred OAuth route; auto-login there needs an explicit authentication port.
+    """
+    parsed = urlsplit(config.base_url)
+    host = f'[{parsed.hostname}]' if ':' in parsed.hostname else parsed.hostname
+    origin = f'{parsed.scheme}://{host}'
+    def endpoint(port):
+        actual = parsed.port if port is None else port
+        return origin + (f':{actual}' if actual is not None else '')
+    api = endpoint(config.dms_port) + (parsed.path or '/proxy_dms')
+    login = (None if parsed.path == '/dms' and config.auth_port is None else
+             endpoint(config.auth_port) + '/proxy_oauth/user/login')
+    return api, login
+
+
 class DmsClient:
     def __init__(self, config, transport=None):
         config.check_scope()
         self.config = config
         self.token = config.token
-        parsed = urlsplit(config.base_url)
-        host = f'[{parsed.hostname}]' if ':' in parsed.hostname else parsed.hostname
-        origin = f'{parsed.scheme}://{host}'
-        self.origin = origin + (f':{parsed.port}' if parsed.port else '')
-        self.auth_origin = origin + (f':{config.auth_port}' if config.auth_port else (f':{parsed.port}' if parsed.port else ''))
-        dms_origin = origin + (f':{config.dms_port}' if config.dms_port else (f':{parsed.port}' if parsed.port else ''))
-        self.api = dms_origin + (parsed.path or '/proxy_dms')
+        self.api, self.login_url = connection_urls(config)
         self.transport = transport or self._http
         self.columns = {}
         self.fields = {}
@@ -181,7 +193,7 @@ class DmsClient:
         return envelope
 
     def normalize_envelope(self, url, envelope):
-        # Verified local DMS list contract: this exact envelope means no rows.
+        # Accept both deployments' list contracts (200/data or exact 202/empty).
         # Never normalize writes, metadata, permission errors or other failures.
         if (url == self.api + '/content/selectContentList' and envelope.get('code') == 202
                 and envelope.get('message') == '数据不存在' and envelope.get('content') == '数据不存在'):
@@ -215,7 +227,7 @@ class DmsClient:
             # Only a definite authentication rejection is replayable.
             with self.auth_lock:
                 if self.token == token:
-                    if not (self.config.username and self.config.password) or (urlsplit(self.config.base_url).path == '/dms' and not self.config.auth_port):
+                    if not (self.config.username and self.config.password) or self.login_url is None:
                         raise DmsError('dms_token_invalid')
                     self._login()
             result = self._request(self.api + path, wire).get('content')
@@ -227,10 +239,10 @@ class DmsClient:
     def _login(self):
         if time.monotonic() < self.auth_retry_at:
             raise DmsError('dms_auth_retry_later')
-        if urlsplit(self.config.base_url).path == '/dms' and not self.config.auth_port:
+        if self.login_url is None:
             raise DmsError('dms_direct_api_requires_token')
         try:
-            result = self._request(self.auth_origin + '/proxy_oauth/user/login', {
+            result = self._request(self.login_url, {
                 'userName': self.config.username, 'password': self.config.password,
                 'clientId': '0', 'serviceId': '0'}, False)
             token = result.get('message')

src/step5_session_analysis/tests/test_dms_connection_modes.py

@@ -0,0 +1,130 @@
+"""The same clients and configuration keys must work in either deployment."""
+
+import copy
+import json
+from pathlib import Path
+from tempfile import TemporaryDirectory
+import unittest
+
+from src.dms_schema import DEFAULT_SCHEMA, names
+from src.step3_question_answer.dms_company import DmsCompanyClient
+from src.step5_session_analysis.dms import DmsClient, DmsError, load_dms_config
+
+
+MODES = (
+    ('http://example.invalid:8080', '', '', '/proxy_dms', 8080, 8080),
+    ('http://example.invalid', '10086', '10081', '/proxy_dms', 10086, 10081),
+    ('http://example.invalid:8080/proxy_dms', '10086', '', '/proxy_dms', 10086, 8080),
+    ('http://example.invalid:8080', '', '10081', '/proxy_dms', 8080, 10081),
+    ('http://example.invalid:8080/dms', '10086', '10081', '/dms', 10086, 10081),
+    ('http://example.invalid:8080/dms', '', '', '/dms', None, 8080),
+)
+
+
+def config_for(base, auth, data, prefixed, company=False, token=''):
+    schema = copy.deepcopy(DEFAULT_SCHEMA)
+    # Deliberately vary every ID and actual field name; never depend on live data.
+    for i, spec in enumerate(schema['columns'].values()):
+        spec['id'] = 9000 + i if prefixed else 8000 + i
+        spec['fields'] = {key: ('c_' + key if prefixed else key) for key in spec['fields']}
+    with TemporaryDirectory() as directory:
+        path = Path(directory)
+        (path / 'schema.json').write_text(json.dumps(schema, ensure_ascii=False), encoding='utf-8')
+        (path / '.env').write_text(
+            f'DMS_BASE_URL={base}\nAUTH_PORT={auth}\nDMS_PORT={data}\n'
+            f'DMS_USERNAME=synthetic\nDMS_PASSWORD=synthetic\nDMS_TOKEN={token}\n', encoding='utf-8')
+        return load_dms_config(path / '.env', {}, schema_path=path / 'schema.json',
+                               scope_names=names(schema, ('basic', 'honor')) if company else None)
+
+
+class ConnectionModeTests(unittest.TestCase):
+    def test_same_config_keys_select_endpoints_and_both_field_layouts(self):
+        for base, auth, data, api_path, auth_port, data_port in MODES:
+            for prefixed in (False, True):
+                for company in (False, True):
+                    with self.subTest(base=base, auth=auth, data=data, prefixed=prefixed, company=company):
+                        config = config_for(base, auth, data, prefixed, company,
+                                            token='supplied' if auth_port is None else '')
+                        calls = []
+                        alias = 'basic' if company else 'sessions'
+                        spec = config.schema['columns'][alias]
+                        logical = 'c_name' if company else 'c_session_id'
+                        actual = spec['fields'][logical]
+                        def transport(url, form, headers):
+                            calls.append((url, copy.deepcopy(form), dict(headers)))
+                            if url.endswith('/user/login'):
+                                return {'code': 200, 'message': 'renewed'}
+                            self.assertEqual(headers['token'], 'supplied' if auth_port is None else 'renewed')
+                            if url.endswith('/selectContentList'):
+                                self.assertEqual(json.loads(form['search'])[0]['field'], actual)
+                                return {'code': 200, 'content': {'count': 1, 'data': [
+                                    {'id': 'synthetic-row', actual: 'synthetic-value'}]}}
+                            self.assertEqual(json.loads(form['content'])[actual], 'synthetic-value')
+                            return {'code': 200, 'content': 'synthetic-row'}
+                        client = (DmsCompanyClient if company else DmsClient)(config, transport)
+                        client.columns[spec['id']] = {}
+                        if auth_port is not None:
+                            client._login()
+                            self.assertEqual(calls[0][0], f'http://example.invalid:{auth_port}/proxy_oauth/user/login')
+                            self.assertNotIn('token', calls[0][2])
+                        rows = client.call('/content/selectContentList', {'columnId': spec['id'],
+                            'search': json.dumps([{'field': logical, 'searchType': 1,
+                                                  'content': {'value': 'synthetic-value'}}])})
+                        self.assertEqual(rows['data'][0][logical], 'synthetic-value')
+                        if not company:
+                            for operation in ('addContent', 'updateContent'):
+                                client.call('/content/' + operation, {'columnId': spec['id'],
+                                    'content': json.dumps({logical: 'synthetic-value'})})
+                        data_calls = [url for url, _, _ in calls if not url.endswith('/user/login')]
+                        self.assertTrue(all(url.startswith(f'http://example.invalid:{data_port}{api_path}/')
+                                            for url in data_calls))
+
+    def test_both_empty_contracts_and_error_isolation_in_each_mode(self):
+        for base, auth, data, _, _, _ in MODES:
+            for company in (False, True):
+                config = config_for(base, auth, data, False, company, token='supplied')
+                identity = config.schema['columns']['basic' if company else 'sessions']['id']
+                for empty in ({'code': 200, 'content': {'count': 0, 'data': []}},
+                              {'code': 202, 'message': '数据不存在', 'content': '数据不存在'}):
+                    client = (DmsCompanyClient if company else DmsClient)(config, lambda *a: empty)
+                    client.columns[identity] = {}
+                    self.assertEqual(client.call('/content/selectContentList', {'columnId': identity}),
+                                     {'count': 0, 'data': []})
+                for invalid in ({'code': 202, 'message': 'different', 'content': '数据不存在'}, {'code': 201}):
+                    client = (DmsCompanyClient if company else DmsClient)(config, lambda *a: invalid)
+                    client.columns[identity] = {}
+                    with self.assertRaises(DmsError):
+                        client.call('/content/selectContentList', {'columnId': identity})
+
+    def test_token_refresh_uses_configured_auth_endpoint_and_is_bounded(self):
+        for base, auth, data, api_path, auth_port, data_port in MODES:
+            for failure in ('envelope', 'http401'):
+                config = config_for(base, auth, data, True, token='expired')
+                calls = []
+                def transport(url, form, headers):
+                    calls.append(url)
+                    if url.endswith('/user/login'):
+                        return {'code': 200, 'message': 'renewed'}
+                    if headers['token'] == 'expired':
+                        if failure == 'http401':
+                            raise DmsError('dms_token_invalid')
+                        return {'code': 212}
+                    return {'code': 200, 'content': 'ok'}
+                client = DmsClient(config, transport)
+                if auth_port is None:
+                    with self.assertRaisesRegex(DmsError, 'dms_token_invalid'):
+                        client.call('/column/getColumnById', {'id': config.sessions_id})
+                    self.assertEqual(len(calls), 1)
+                else:
+                    self.assertEqual(client.call('/column/getColumnById', {'id': config.sessions_id}), 'ok')
+                    self.assertEqual(len(calls), 3)
+                    self.assertEqual(calls[1], f'http://example.invalid:{auth_port}/proxy_oauth/user/login')
+                    self.assertEqual(calls[0], calls[2])
+
+    def test_direct_api_without_auth_port_never_guesses_a_login_address(self):
+        config = config_for('https://example.invalid/dms', '', '', False)
+        calls = []
+        client = DmsClient(config, lambda *args: calls.append(args))
+        with self.assertRaisesRegex(DmsError, 'dms_direct_api_requires_token'):
+            client.connect()
+        self.assertEqual(calls, [])