Xero认证返回401 Unauthorized，交易查询脚本无法运行

我正在开发一个脚本，通过包含交易参考号的CSV文件查询Xero中的交易，但运行代码时一直收到未授权响应。已查阅Xero官方文档，确认完成了所有认证步骤，但问题仍未解决。

以下是我的代码：

import requests
import csv
from urllib.parse import urlencode
from flask import Flask, request

# Set up OAuth 2.0 credentials
client_id = 'REDACTED'
client_secret = 'REDACTED'
redirect_uri = 'REDACTED'
authorization_url = 'https://login.xero.com/identity/connect/authorize'
token_url = 'https://identity.xero.com/connect/token'
api_base_url = 'https://api.xero.com/api.xro/2.0'

# Generate the authorization URL
def generate_authorization_url():
    params = {
        'response_type': 'code',
        'client_id': client_id,
        'redirect_uri': redirect_uri,
        'scope': 'openid profile email accounting.transactions offline_access',

    }
    query_string = urlencode(params)
    return f'{authorization_url}?{query_string}'

# Retrieve authorization code from callback URL
def retrieve_authorization_code(callback_url):
    authorization_code = None
    if 'code' in callback_url:
        authorization_code = callback_url.split('code=')[1].split('&')[0]
    return authorization_code

# Exchange authorization code for access token
def exchange_authorization_code(authorization_code):
    data = {
        'grant_type': 'authorization_code',
        'code': authorization_code,
        'client_id': client_id,
        'client_secret': client_secret,
        'redirect_uri': redirect_uri
    }
    response = requests.post(token_url, data=data)
    if response.status_code == 200:
        return response.json().get('access_token')
    else:
        raise Exception('Error exchanging authorization code for access token.')

# Search for transactions using the CSV file and the "Details" column
def search_transactions(access_token, csv_file, column_name):
    transactions = []
    not_found_transactions = []

    # Load transactions from CSV file
    with open(csv_file, 'r') as file:
        csv_reader = csv.DictReader(file)
        csv_data = list(csv_reader)

    # Search for transactions in Xero
    headers = {
        'Authorization': f'Bearer {access_token}',
        'Accept': 'application/json',
        'Content-Type': 'application/json'
    }
    for row in csv_data:
        details = row[column_name]
        # Search transactions by details
        params = {
            'where': f'Details=="{details}"'
        }
        response = requests.get(f'{api_base_url}/BankTransactions', headers=headers, params=params)
        if response.status_code == 200:
            search_result = response.json().get('BankTransactions')
            if search_result:
                transactions.extend(search_result)
            else:
                not_found_transactions.append(row)
        else:
            print(f'Error searching transactions in Xero. Status Code: {response.status_code}')
            print(f'Response Content: {response.content}')
            raise Exception('Error searching transactions in Xero.')

    return transactions, not_found_transactions

# Start a simple Flask web service on http://localhost:3000
app = Flask(__name__)

@app.route('/')
def index():
    return 'Web service is running.'

@app.route('/callback')
def callback():
    callback_url = request.url
    authorization_code = retrieve_authorization_code(callback_url)
    if authorization_code:
        # Exchange authorization code for access token
        access_token = exchange_authorization_code(authorization_code)

        # Search for transactions using the CSV file and the "Details" column
        csv_file = 'transactions.csv'
        column_name = 'Details'
        transactions, not_found_transactions = search_transactions(access_token, csv_file, column_name)

        # Print matched transactions
        print('Matched Transactions:')
        for transaction in transactions:
            print(transaction)

        # Print not found transactions
        print('Not Found Transactions:')
        for transaction in not_found_transactions:
            print(transaction)

    return 'Authorization code retrieved successfully.'

if __name__ == '__main__':
    # Generate the authorization URL
    auth_url = generate_authorization_url()
    print(f'Please visit the following URL to authorize the app: {auth_url}')

    # Run the Flask web service on http://localhost:3000
    app.run('localhost', 3000)

核心排查点及修复建议 #

• 缺少租户ID请求头：Xero API要求所有数据请求必须携带Xero-tenant-id头，当前代码完全遗漏了这个关键参数。在交换授权码时，Xero会返回tenant_id，需保存并添加到请求头：

  # 改造token交换函数，返回tenant_id
  def exchange_authorization_code(authorization_code):
      data = {
          'grant_type': 'authorization_code',
          'code': authorization_code,
          'client_id': client_id,
          'client_secret': client_secret,
          'redirect_uri': redirect_uri
      }
      response = requests.post(token_url, data=data)
      if response.status_code == 200:
          token_data = response.json()
          return token_data.get('access_token'), token_data.get('tenant_id')
      else:
          raise Exception(f'Token交换失败: {response.content}')
  
  # 搜索函数添加tenant_id参数，更新请求头
  def search_transactions(access_token, tenant_id, csv_file, column_name):
      headers = {
          'Authorization': f'Bearer {access_token}',
          'Accept': 'application/json',
          'Content-Type': 'application/json',
          'Xero-tenant-id': tenant_id
      }
      # ... 其余代码不变
  

• 授权码获取方式不可靠：用字符串分割解析授权码容易出错，改用Flask内置的参数解析：

  def retrieve_authorization_code():
      return request.args.get('code')
  

• Scope权限验证：确认用户授权时未拒绝accounting.transactions权限，可在授权后通过Xero的身份端点检查token包含的权限范围。
• 查询参数编码：直接拼接where条件可能因特殊字符导致语法错误，建议对details进行URL编码：

  from urllib.parse import quote
  params = {
      'where': f'Details=="{quote(details)}"'
  }
  

内容的提问来源于stack exchange，提问作者MichaelY