# 背景
由于celery最小只能到分钟级别，但是日常有需要每隔10秒同步一遍获取远程系统能力

# 实现

```python3
""" Auto reload celery runner script
Usage: python manage.py sync_perms
See: https://stackoverflow.com/a/49166246/2544762
"""
import signal
import traceback
from datetime import datetime
from logging import getLogger
from time import sleep
from typing import Dict

from django.apps import apps
from django.conf import settings
from django.core.management.base import BaseCommand

from mobile.tasks import job_watch_sonic_devices

logger = getLogger(__name__)


class Command(BaseCommand):
    def handle(self, *args, **options):
        # 退出标记
        flag_terminate = False

        # 退出信号处理（Ctrl + C），使得可以正常结束不用 kill
        def signal_handler(signal, frame):
            nonlocal flag_terminate
            flag_terminate = True

        signal.signal(signal.SIGINT, signal_handler)
        signal.signal(signal.SIGTERM, signal_handler)

        # 死循环轮询定时任务
        while not flag_terminate:
            logger.info(f'定时任务开始执行：{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
            try:
                job_watch_sonic_devices()
            except Exception as e:
                print(f'watch sonic e: {e}')
                traceback.print_exception(e)
            # 整体执行时间需要2秒，这里设置8秒，基本10秒内一个轮回
            sleep(8)

```
## 1、信号处理

```python3
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
```
捕获 Ctrl + C (SIGINT) 和 终止信号 (SIGTERM)。

使用 nonlocal flag_terminate 让信号处理函数能修改外部变量，确保可以优雅退出循环，而不是被强制 kill。

优点：

运行中按 Ctrl+C 可以安全结束（会在下个循环退出）。

避免直接 kill -9 导致状态不一致。


```python3
flag_terminate = False

def signal_handler(signal, frame):
    nonlocal flag_terminate
    flag_terminate = True

```
### 底层流程（信号机制是怎么帮忙的）：
当用户 Ctrl+C（SIGINT）或 kill（SIGTERM）时，不是立刻杀掉进程，而是执行 signal_handler：

把 flag_terminate 设成 True。

主循环每轮都会检查 flag_terminate：

如果 True，退出循环，任务结束前可以优雅清理（比如保存状态、断开连接）。

这样：

正在执行的那一轮任务会先跑完。

然后在下一轮循环检查到退出标志，再干净地退出。

### 简单理解：

信号处理就是给死循环任务加了一个“安全刹车”，让你按下 Ctrl+C 时，不是直接翻车，而是慢慢停下来。

### 是否同步？
如果只看 Python 解释器层面，当收到信号时，信号处理函数（signal_handler）的执行是同步的。

解释器会在执行字节码的“安全点”检测是否有挂起的信号：

如果有，就暂停当前的 Python 代码执行；

同步调用你注册的 signal.signal(...) 处理函数；

处理函数执行完后，回到原来的位置继续执行

## 2、技术要点总结

- 自定义 Django 管理命令：可直接用 python manage.py <command> 启动，不依赖外部调度器。
- 信号安全退出：通过 signal.signal + 外部变量标志实现优雅退出。
- 异常隔离：任务出错不会中断整个进程。
- 简单定时调度：用 sleep() 模拟固定周期调度（适合简单场景，不适合精确到秒的生产任务）。
- 日志追踪：方便监控执行情况。
- 可替代 Celery beat：适合轻量任务或测试环境，但在高并发/分布式场景下不如 Celery 稳定

django中增加manage.py拉起命令及信号机制

# 背景
日常多表查询时，容易引入脏数据，需要删除多表中脏数据，减少直接操作DB的工作量

# model层代码分析
假设有表如下：

TestTaskPackagePluginApplianceRel 中间表，挂钩插件包和插件设备

TestTaskPackage 插件包

TestTaskPluginAppliance 插件设备

```python3
from django.db import transaction
class TestTaskPackage(UUIDModel, DatedModel):
    ...
    
    def delete_with_related(self):
            """删除当前 TestTaskPackage 及其关联的 TestTaskPluginAppliance 和中间表记录"""
            with transaction.atomic():

                # 先查出来关联TestTaskPluginAppliance的记录并强制评估为列表
                appliance_ids = TestTaskPackagePluginApplianceRel.objects.filter(test_task_package=self).values_list(
                    'test_task_plugin_appliance_id', flat=True
                )
                appliances = list(TestTaskPluginAppliance.objects.filter(id__in=appliance_ids))

                logger.info(f'appliances: {appliances}')

                # 删除中间表中的记录
                rels = TestTaskPackagePluginApplianceRel.objects.filter(test_task_package=self)
                for rel in rels:
                    logger.info(f"Deleting relation: {rel}")
                    rel.delete()

                # 删除关联的 TestTaskPluginAppliance
                # logger.info(f'appliances after: {appliances}')
                for appliance in appliances:
                    logger.info(f"Deleting appliance: {appliance}")
                    appliance.delete()

                # 删除当前 TestTaskPackage
                logger.info(f"Deleting TestTaskPackage: {self}")
                self.delete()

```
## 1、事务管理 (transaction.atomic)

```python3
with transaction.atomic():
```
作用：把代码块中的数据库操作放在同一个事务里执行。

好处：

如果中间任何一步抛异常，所有已执行的 SQL 都会回滚，保证数据一致性。

避免出现删除了一部分关联记录但没有完全清理干净的情况

## 2、多对多/外键关联查询 (filter + values_list)

```python3
appliance_ids = TestTaskPackagePluginApplianceRel.objects.filter(
    test_task_package=self
).values_list('test_task_plugin_appliance_id', flat=True)

```
filter：筛选出中间表 TestTaskPackagePluginApplianceRel 中，关联到当前 TestTaskPackage 的记录。

values_list(..., flat=True)：直接取出字段值列表（只要一个字段时 flat=True 才能得到一维列表）。

这里的用途：先获取所有相关的 TestTaskPluginAppliance ID，方便后面批量删除

## 3、强制转换到内存 QuerySet (list(...))

```python3
appliances = list(TestTaskPluginAppliance.objects.filter(id__in=appliance_ids))
```
jango 的 QuerySet 是惰性执行的（lazy evaluation）。

用 list(...) 会立刻执行 SQL 查询，把结果加载到内存中。

这样做的原因：

后续删除中间表记录时，这些对象仍然能在内存中访问，避免因级联删除或 QuerySet 变化导致数据不一致。

# views层

```python3
class TestTaskPackageViewSet(viewsets.ModelViewSet):
    queryset = m.TestTaskPackage.objects.all()
    serializer_class = s.TestTaskPackageSerializer
    filter_fields = '__all__'
    ordering = ['-pk']
    permission_classes = [ActionBasedPermission]
    action_permissions = {
        'list,retrieve': IsMember,
        'create': IsMember,
        'update,partial_update,destroy,delete_with_related':
            IsMember |
            UserWithCondition.build(lambda u, o: u == o.author),
    }
    
    @action(methods=['POST'], detail=False)
    def delete_with_related(self, request):
        """删除指定的 TestTaskPackage 及其关联的记录"""
        from rest_framework import status
        try:
            test_task_id = request.data.get('test_task_id')
            logger.info(f'test_task_id: {test_task_id}')
            test_task_package = (m.TestTaskPackage.objects.filter(test_task_id=test_task_id,
                                                                  package_type=m.TestTaskPackage.TYPE_PLUGIN_PACKAGE).
                                 order_by('date_created').first())
            logger.info(f'test_task_package: {test_task_package}')
            # 检查 test_task_package 是否为空
            if not test_task_package:
                return Response(
                    {'status': 'info', 'message': '当前没有关联任何插件包，跳过删除操作。'},
                    status=status.HTTP_200_OK
                )
            test_task_package.delete_with_related()
            return Response(
                {'status': 'success', 'message': 'TestTaskPackage and related records deleted successfully.'},
                status=status.HTTP_204_NO_CONTENT)
        except Exception as e:
            return Response({'status': 'error', 'message': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
```

# 测试层分析

```python3
from XXX.models import GrpcMemberProxy

from django.test import TestCase
from plugin_test import models as m
import uuid
from datetime import datetime


class TestTaskPackageModelTest(TestCase):

    def setUp(self):
        # 创建一个 TestTaskPackage 实例
        author = GrpcMemberProxy(username='abc')
        self.test_task_package = m.TestTaskPackage.objects.create(
            id=uuid.uuid4(),
            test_task_id=uuid.uuid4(),
            mobile_device_id=123456789,
            file_md5='dummy_md5',
            author=author
        )

        # 创建一个 TestTaskPluginAppliance 实例
        self.test_task_plugin_appliance = m.TestTaskPluginAppliance.objects.create(
            name='Test Appliance',
            appliance_code='dummy_code',
            sn='dummy_sn',
            bt_mac='dummy_bt_mac',
            enterprise_name='Dummy Enterprise',
            appliance_type='Dummy Type',
            enterprise_code='dummy_ent_code',
            cat_code='dummy_cat_code',
            sn8='dummy_sn8',
            develop_group_name='Dummy Group',
            product_code='dummy_product_code',
            wifi_version='dummy_wifi_version',
            register_time=datetime.now(),  # 提供当前时间作为注册时间
            a0='dummy_a0',
            prd_name='Dummy Product',
            prd_id='dummy_prd_id',
            model_type='dummy_model_type',
            access_method='dummy_access_method',
            source='dummy_source',
            product_code_in_app_svc='dummy_product_code_in_app_svc',
            smart_product_id=1,
            prd_info={},  # 提供一个空字典作为默认值
            prd_id_on_spid='dummy_prd_id_on_spid'
        )

        # 创建一个中间表记录
        m.TestTaskPackagePluginApplianceRel.objects.create(
            test_task_package=self.test_task_package,
            test_task_plugin_appliance=self.test_task_plugin_appliance
        )

    def test_delete_with_related(self):
        # 确保中间表记录存在
        self.assertEqual(m.TestTaskPackagePluginApplianceRel.objects.count(), 1)
        # 确保 TestTaskPluginAppliance 记录存在
        self.assertEqual(m.TestTaskPluginAppliance.objects.count(), 1)

        # 调用 delete_with_related 方法
        self.test_task_package.delete_with_related()

        # 检查所有相关记录是否被删除
        self.assertEqual(m.TestTaskPackage.objects.filter(id=self.test_task_package.id).count(), 0)
        self.assertEqual(m.TestTaskPluginAppliance.objects.filter(id=self.test_task_plugin_appliance.id).count(), 0)
        self.assertEqual(m.TestTaskPackagePluginApplianceRel.objects.count(), 0)

```

## 1、jango.test.TestCase
```python3
class TestTaskPackageModelTest(TestCase):
```
继承自 django.test.TestCase，意味着：

每个 test_ 方法运行前都会新建一个测试数据库（事务隔离）。

每个测试方法运行后会回滚数据库，不影响其他测试。

运行速度快且测试之间数据互不干扰。

## 2、setUp 阶段

```python3
def setUp(self):
```
setUp 会在每个测试方法执行前自动运行，用来准备测试数据。

## 3、中间关联表
```python3
m.TestTaskPackagePluginApplianceRel.objects.create(
    test_task_package=self.test_task_package,
    test_task_plugin_appliance=self.test_task_plugin_appliance
)
```
建立 TestTaskPackage 和 TestTaskPluginAppliance 的关联关系。

这相当于多对多关系的“中间表”。

数据库里现在三张表各有一条相关记录。

## 4、流程总结
1）准备数据（setUp）

插入一条包记录、一条设备记录、一条中间表记录。

2）验证初始状态

确认数据已按预期插入。

3）执行删除方法

调用 delete_with_related()。

4）验证最终状态

所有关联数据都被删光。

5）测试结束回滚

Django 自动清空测试数据库，准备下一个测试

# vue2前端要点
template
```js
<i-button class="btn" v-if="me.is_superuser" @click="clearTaskRelated()" :type="'error'" :disabled="!me.is_superuser">脏数据清除</i-button>
```
    
script
```js
// 脏数据清除（物模型插件任务/关联表/插件设备表）
    async clearTaskRelated() {
      const vm = this;

      // 显示确认弹窗
      await vm.$confirm('您确定要删除该任务及其关联的数据吗？此操作不可撤销!!!')

      try {

        // 调用后端删除接口
        const deleteResp = await vm.api('test_task_package').showLoading(false).post({
          action: 'delete_with_related'
        }, {
          test_task_id: vm.$route.query.task
        });

        // 检查删除操作的响应
        if (deleteResp.status === 204) {
          vm.$Message.success({
            content: '任务及其关联数据已成功删除。',
            duration: 3
          });
        } else {
          vm.$Message.error({
            content: deleteResp.data.message,
            duration: 5
          });
        }
      } catch (error) {
        console.error('Error:', deleteResp.data.message);
        vm.$Message.error({
          content: '操作失败，请检查网络连接或稍后重试。',
          duration: 5
        });
      }
    }
```
## 1、调用后端 delete_with_related Action：

第一个参数 { action: 'delete_with_related' } → 这是后端的 DRF Action 调用方式。

第二个参数 { test_task_id: vm.$route.query.task } → 把任务 ID 传给后端。

showLoading(false)：

禁用全局 loading 动画。

设计要点：

test_task_id 从 URL query 参数中获取（this.$route.query.task），无需额外输入。

API 请求是 POST，符合 DRF @action(methods=['POST']) 要求。

## 2、技术要点总结
1）操作前确认：用弹窗防止误删，明确提示危险操作。

2）API 调用模式：

链式调用 (vm.api(...).showLoading(false).post(...))。

动作参数 { action: 'delete_with_related' } 对应 DRF 自定义 Action。

3）参数传递：test_task_id 来自 URL query，确保接口参数和后端匹配。

4）状态码处理：

204 表示删除成功。

其他状态用后端返回的 message 提示用户。

5、异常处理：

捕获请求错误，防止 UI 卡死。

提示用户并记录日志。


django事务处理删除事件前后端实现要点

# 背景
前后端开发时，日常通常需要nginx配置转发，使得同一个网址可以直接打开访问

# nginx配置

```bash
virtual-machine:/etc/nginx/conf.d$ cat django.conf
server {
 
    listen 80;
    listen [::]:80;
 
    server_name localhost app.*;
 
    index index.html;
    client_max_body_size 200m;
    keepalive_timeout 120s;
     
    location @webpack8080 {
        proxy_redirect off;
        proxy_pass http://127.0.0.1:8080;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        client_max_body_size 20m;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
 
    location @django {
        proxy_redirect off;
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        client_max_body_size 20m;
    }
 
    location / {
        try_files index.html @webpack8080;
    }
 
    location ~ ^/(auth|api|api-auth|admin|_admin|member|media|static|wechat_oauth)(/|$) {
        try_files index.html @django;
    }
#    
#    location ~ ^/service_proxy/apphub/(?<service>[\w\d_-]+) {
#        rewrite ^/service_proxy/apphub/([\w\d_-]+)/(.+)$ /$2 break;
#        proxy_pass http://10.23.105.108;
#        proxy_set_header Host $service.sit-iottest.aaa.com;
#    } 
   location ~ ^/service_proxy/apphub/(.+)$ {
      rewrite ^/service_proxy/apphub/(.+)$ /$1 break;
      proxy_pass http://10.23.105.108;
      proxy_set_header Host apphub.sit-iottest.aaa.com;
   }
}

```

前后端Nginx配置

# 背景
很多算法简单等级的题目，不需要比较复杂的思路，直接就可以正常思路解决（常见于日常编码逻辑固化后），这里只需要知道常规的高频使用的点即可（默认Python3刷题）

# 列表
```bash
# 删除某个值
>>> lst = [1,2,3]
>>> lst.remove(2)
>>> lst
[1, 3]

# 逆向遍历
for j in range(n - 1, 0, -1):
  print(j)
#这种情况没有遍历到下标为 0 的元素，这是一个常见的 range() 使用误区
# 正确使用：Python 的 range(start, stop, step) 是左闭右开的
for j in range(n - 1, -1, -1):

# 下标联动的枚举
>>> nums = [2,1,3,3]
>>> [(num, i) for i, num in enumerate(nums)]
[(2, 0), (1, 1), (3, 2), (3, 3)]
```

# 字符串

```bash
# 前后缀判断 prefix suffix
>>> 'abab'.startswith('ab')
True
>>> 'abab'.endswith('ab')
True
# 纯数字、纯字母、纯数字和字母组合
>> '11'.isdigit()
True
>>> 
>>> 'a1'.isalnum()
True
>>> 
>>> 'aa'.isalpha()
True

# join()用法， 序列中的元素连接成一个字符串 '连接符'.join(可迭代对象)
>>> '-'.join(['a', 'b', 'c'])
'a-b-c'
>>> ''.join(['a', 'b', 'c'])
'abc'

# 所有子串
>> s = "10101"
>>> [s[i:j] for i in range(len(s)) for j in range(i+1, len(s) + 1)]
['1', '10', '101', '1010', '10101', '0', '01', '010', '0101', '1', '10', '101', '0', '01', '1']
```

# 集合

```bash
>>> {'a', 'b'} in {'a', 'b'}  # 注意！！！ 集合 {'b', 'a'} 是否是集合 {'b', 'a'} 的一个元素，这个是不能直接这么用
False
>>> {'a', 'b'}.issubset({'a', 'b'})
True
>>> {'a', 'b'}.issubset({'b', 'a'})  # 集合是没有顺序的概念
True
```

# 字符
字符运算及转换
```bash
>>> ord('a')
97
>>> chr(97)
'a'
>>> ord('a') - ord('A')
32
```


# string库
string 是一个内置标准库

字符范围判断
```bash
>>> import string
>>> string.ascii_letters
'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
>>> string.ascii_lowercase
'abcdefghijklmnopqrstuvwxyz'
>>> 
>>> string.ascii_uppercase
```

# map方法
map() 是一个内置函数，非原地修改。
将一个函数作用到可迭代对象的每个元素上，并返回一个新的迭代器（map 对象）

map(function, iterable)

```bash
>> def func(x: int) -> int:
...     return x + x
... 
>>> new_lst = map(func, lst)
>>> 
>>> new_lst
<map object at 0x7035210a0910>
>>> list(new_lst)
[2, 4, 6]
```
# zip方法
 Python 的一个内置函数，用来将多个可迭代对象“打包”成一个个元组，返回一个可迭代对象（zip 对象）
```bash
>>> a = [1, 2, 3]
... b = ['a', 'b', 'c']
... 
... z = zip(a, b)
... print(list(z))  # 转成列表查看结果
... 
[(1, 'a'), (2, 'b'), (3, 'c')]

# 高频用法
>>> list(zip(a, a[1:]))
[(1, 2), (2, 3)]
```
# itertools
迭代器函数工具集，用于高效地操作可迭代对象,常用于遍历、组合、过滤

```bash
# 特定组合全排列
>>> import itertools
>>> itertools.permutations('ABC')
<itertools.permutations object at 0x72613a04d080>
>>> list(itertools.permutations('ABC'))
[('A', 'B', 'C'), ('A', 'C', 'B'), ('B', 'A', 'C'), ('B', 'C', 'A'), ('C', 'A', 'B'), ('C', 'B', 'A')]
>>> list(itertools.permutations([1,2,3]))
[(1, 2, 3), (1, 3, 2), (2, 1, 3), (2, 3, 1), (3, 1, 2), (3, 2, 1)]

# 三选二
>>> list(itertools.combinations([1,2,3], 2))
[(1, 2), (1, 3), (2, 3)]

# 笛卡尔积
>>> list(itertools.product('ABCD', '123'))
[('A', '1'), ('A', '2'), ('A', '3'), ('B', '1'), ('B', '2'), ('B', '3'), ('C', '1'), ('C', '2'), ('C', '3'), ('D', '1'), ('D', '2'), ('D', '3')]
```

# collections
高级容器数据类型库，快速计数，队列，默认值字典等等

```bash
# 快速统计，不用自己写字典了，刷算法题时很有用
>>> from collections import Counter
... 
>>> words = ['aa', 'bb', 'aa', 'a', 'b', 'bb']
>>> Counter(words)
Counter({'aa': 2, 'bb': 2, 'a': 1, 'b': 1})
>>> Counter(words).most_common(3)  #前3复合情况
[('aa', 2), ('bb', 2), ('a', 1)]

>>> from collections import defaultdict
... 
>>> dd = defaultdict(int)
... 
>>> dd
defaultdict(<class 'int'>, {})
>>> 
>>> dd['a'] += 1
... 
>>> 
>>> dd
defaultdict(<class 'int'>, {'a': 1})
```

# 时间间隔及交集
datetime 是Python 内置的处理日期和时间的标准库模块

```bash
# 字符串转日期
>>> fmt = "%m-%d"
>>> datetime.strptime('08-10', fmt)
datetime.datetime(1900, 8, 10, 0, 0)

# 日期转字符串
>>> datetime.strptime('08-10', fmt).strftime('%Y-%m-%d')
'1900-08-10'

# 判断两个日期区间是否存在交集
# 字符串转日期对象
fmt = "%Y-%m-%d"
s1, e1 = datetime.strptime(start1, fmt), datetime.strptime(end1, fmt)
s2, e2 = datetime.strptime(start2, fmt), datetime.strptime(end2, fmt)

# 计算交集
start = max(s1, s2)
end = min(e1, e2)

# 判断是否有交集
if start <= end:
    return start.strftime(fmt), end.strftime(fmt)
else:
    return None  # 无交集

# 交集天数
(end-start).days + 1
```

# 最大公约数

```bash
>>> def gcd(x: int, y: int) -> int:
...     """最大公约数"""
...     while y % x != 0:
...         x,y = y%x, x
...     return x
... 
>>> gcd(10, 30)
10
>>> 
>>> gcd(15, 5)
5
```
# 最小公倍数

```bash
>>> def lcm(x: int, y: int) -> int:
...     """最小公倍数"""
...     return x * y // gcd(x, y)
... 
>>> lcm(15, 20)
60
```

# 素数

```bash
>>> def is_prime(num: int) -> bool:
...     """判断一个数是否为素数"""
...     for i in range(2, int(num ** 0.5)+1):
...         if num % i == 0:
...             return False
...     return True
... 
>>> is_prime(5)
True
```

# 整数数位操作技巧 // %

```bash
# 整数反转
>>> num
1357
>>> while num > 0:
...     reversed_num = reversed_num * 10 +num % 10
...     num //= 10
... 
>>> reversed_num
7531
```

# 排序
注意：sorted()为非原地，.sort()为**列表**专属的原地排序！！！
```bash
# 原地排序
>>> lst = [1,5,3]
>>> lst.sort()
>>> lst
[1, 3, 5]

# 非原地排序
>>> result = {1: 2, 3: 4, 5: 3}
>>> sorted(result)
[1, 3, 5]
>>> result
{1: 2, 3: 4, 5: 3}
# 根据字典的value值排序
>>> sorted(result, key=lambda x: result[x])
[1, 5, 3]
```

# 双指针

```python3
# 有序情况下，考虑左右指针移动
# from typing import List
#
# class Solution:
#     def twoSum(self, numbers: List[int], target: int) -> List[int]:
#         num_dict = {}
#         for i, num in enumerate(numbers):
#             complement = target - num
#             if complement in num_dict:
#                 return [num_dict[complement], i]
#             num_dict[num] = i
#         return []
#
#
# if __name__ == '__main__':
#     numbers = [2,3,4]
#     target = 6
#     solution = Solution()
#     result = solution.twoSum(numbers, target)
#     print(result)  # Output: [1, 3]

from typing import List


class Solution:
    def twoSum(self, numbers: List[int], target: int) -> List[int]:
        left, right = 0, len(numbers) - 1
        while left < right:
            temp = numbers[left] + numbers[right]
            if temp == target:
                return [left, right]  # 或者 [left + 1, right + 1]
            elif temp < target:
                left += 1
            else:
                right -= 1
        return []  # 没找到就返回空列表


if __name__ == '__main__':
    numbers = [2, 3, 4]
    target = 6
    solution = Solution()
    result = solution.twoSum(numbers, target)
    print(result)  # Output: [1, 2]

```
补充，如果无序情况，可以考虑使用哈希表法

```python3
num_dict = {}
for i, num in enumerate(numbers):
    complement = target - num
    if complement in num_dict:
        return [num_dict[complement], i]
    num_dict[num] = i
```

# 剪枝替换思路:缓存
已知以下代码执行很慢（数据量上来时）
```python3
def fib(num):
    if num in (1, 2):
        return 1
    return fib(num - 1) + fib(num - 2)
```

借助缓存实现

```python3
from functools import lru_cache


@lru_cache()
def fib(num):
    if num in (1, 2):
        return 1
    return fib(num - 1) + fib(num - 2)


for n in range(1, 121):
    print(f'{n}: {fib(n)}')
```

# 遍历获取下标
默认常规遍历获取下标，时间复杂度为O(n*n)

```python3
lst = []
for num in arr:  # O(n)
    lst.append(sorted_arr.index(num)+1)  # O(n)
return lst
```
借助空间(空间换时间)

```python3
rank = {num: i + 1 for i, num in enumerate(sorted_arr)}  # 构建映射，O(n)
return [rank[num] for num in arr]  # 替换为排名，O(n)
```

算法暴力刷题高频使用点

# 排序算法比较

```bash
| 排序算法 | 最好时间复杂度    | 最坏时间复杂度    | 平均时间复杂度         | 空间复杂度    | 是否稳定  | 核心思想            |
| ---- | ---------- | ---------- | --------------- | -------- | ----- | --------------- |
| 冒泡排序 | O(n)       | O(n²)      | O(n²)           | O(1)     | ✅ 稳定  | 相邻比较，大的冒后面      |
| 选择排序 | O(n²)      | O(n²)      | O(n²)           | O(1)     | ❌ 不稳定 | 每轮选最小，放前面       |
| 插入排序 | O(n)       | O(n²)      | O(n²)           | O(1)     | ✅ 稳定  | 当前元素插入已排好序列     |
| 快速排序 | O(n log n) | O(n²)      | O(n log n)      | O(log n) | ❌ 不稳定 | 分治 + 找基准 + 左右排序 |
| 归并排序 | O(n log n) | O(n log n) | O(n log n)      | O(n)     | ✅ 稳定  | 分治 + 合并有序子序列    |
| 希尔排序 | O(n log n) | O(n²)      | O(n^1.3)\~O(n²) | O(1)     | ❌ 不稳定 | 插入排序的优化（分组）     |
| 堆排序  | O(n log n) | O(n log n) | O(n log n)      | O(1)     | ❌ 不稳定 | 构建大顶堆/小顶堆       |
| 计数排序 | O(n + k)   | O(n + k)   | O(n + k)        | O(n + k) | ✅ 稳定  | 用数组记录频次         |
| 桶排序  | O(n + k)   | O(n²)      | O(n + k)        | O(n + k) | ✅ 稳定  | 分桶 + 各桶分别排序     |
| 基数排序 | O(nk)      | O(nk)      | O(nk)           | O(n + k) | ✅ 稳定  | 按位排序（从低到高）      |

```



# python自带排序
Python 默认的排序使用的是 Timsort（提姆排序） 算法，

Timsort = Merge Sort（归并排序） + Insertion Sort（插入排序） 的优化组合


```bash
| 特性          | 说明                       |
| ----------- | ------------------------ |
| **稳定排序**    | 相等元素顺序不变 ✅               |
| **时间复杂度**   | 最好 O(n)，平均/最坏 O(n log n) |
| **空间复杂度**   | O(n) 辅助空间                |
| **优化小数组**   | 小段用插入排序                  |
| **优化已排序数据** | 已部分有序数据时性能更好（比快速排序快）     |

```



# 冒泡排序
思路：

```bash
相邻元素比较，把大的交换到后面

每轮把最大值“冒泡”到末尾
```
```bash
>> def bubble_sort(lst: list) -> list:
...     """冒泡排序
...     exam:
...         origin: [10, 8, 9, 6]
...         思路：相邻比较，大的往后放
...         第一轮：
...         first: [8, 10, 9, 6]
...         second: [8, 9, 10, 6]
...         third: [8, 9, 6, 10]
...         第二轮： [9, 6, 10]依次思路去比较
...         ...
...     """
...     n = len(lst)
...     # 外层控制轮数
...     for i in range(n):
...         # 内层控制相邻比较次数
...         for j in range(0, n-i-1):
...             if lst[j] > lst[j+1]:
...                 lst[j], lst[j+1] = lst[j+1], lst[j]    
```

简单优化：加入交换标记，如果一轮下来啥都没有交换，默认已经为交换完成，后续不需要再比较

```bash
>>> def bubble_sort(lst: list) -> list:
...     """冒泡排序
...     exam:
...         origin: [10, 8, 9, 6]
...         思路：相邻比较，大的往后放
...         第一轮：
...         first: [8, 10, 9, 6]
...         second: [8, 9, 10, 6]
...         third: [8, 9, 6, 10]
...         第二轮： [9, 6, 10]依次思路去比较
...         ...
...     """
...     n = len(lst)
...     # 外层控制轮数
...     for i in range(n):
...         swapper = False  # 标记是否有交换
...         # 内层控制相邻比较次数
...         for j in range(0, n-i-1):
...             if lst[j] > lst[j+1]:
...                 lst[j], lst[j+1] = lst[j+1], lst[j]    
...                 swapper = True
...         if not swapper:
...             break
```

# 插入排序
思路：
```bash
当前元素插入到左边已排序序列中

类似打扑克牌理牌
```

```bash
>>> def insertion_sort(arr: list):
...     """插入排序： 当前元素插入到左边已排序序列中
...     将第一个元素看作已排序部分
...     从第二个元素开始，向前找到它应插入的位置
...     插入后，前 i 个元素就是有序的
...     重复直到最后一个元素
...     """
...     for i in range(1, len(arr)):
...         key = arr[i]
...         j = i - 1
...         # 向左比较并移动元素
...         while j>=0 and arr[j] > key:
...             arr[j + 1] = arr[j]
...             j -= 1
...         arr[j + 1] = key
... 
>>> arr = [7, 3, 5, 2]
>>> insertion_sort(arr)
>>> arr
[2, 3, 5, 7]
```

# 快排
思路：每轮通过基准值，分拆成3块，比基准值小的部分，基准值，比基准值大的部分，再依次对基准值小的部分、大的部分，做类似的分拆，直到剩1个值或为空时返回（这里之所以必须要小于等于1，是因为如果以2个值为基准时，这两个是可能乱序，只剩1个或0个，则返回必然是有序的）

例子：[3,1,2,5,6,7]
- 1、借助递归思路，最外层递归实现为return [1,2] + [3] + [5,6,7]
- 2、再往里一层left_lst=[1,2]则是由[] + [1] + [2]
- 3、同层right_lst=[5,6,7]则是由[] + [5] + [6,7]，这里的[6,7]则是由继续再往里一层递归返回[6,7]=[] + [6] + [7]

**时间复杂度：**O(n*n)，平均时间复杂度为O(nlogn)

```bash
def quickSorted(lst):

    #base
    if len(lst) <= 1:
        return lst

    pivot = lst[0]#基准值

    left_lst = quickSorted([item for item in lst[1:] if item<pivot])
    right_lst = quickSorted([item for item in lst[1:] if item>=pivot])
    return left_lst + [pivot] + right_lst
```

# 归并排序

```bash
>>> def merge_sort(lst: list) -> list:
...     """分解（Divide）：
... 
... 把数组从中间分成两半
... 
... 对每一半递归进行归并排序
... 
... 解决（Conquer）：
... 
... 每次递归到长度为 1，认为是有序的
... 
... 合并（Merge）：
... 
... 把两个有序子数组合并成一个有序大数组"""
...     if len(lst) <= 1:
...         return lst
...     # half
...     mid = len(lst) // 2
...     left = merge_sort(lst[:mid])
...     right = merge_sort(lst[mid:])
...     # merge
...     return merge(left, right)
... 
>>> def merge(left: list, right: list) -> list:
...     result = []
...     i = j  = 0  # 左右指针，分别指向left right列表
...     # 比较左右两边元素大小，依次放入result
...     while i<len(left) and j<len(right):
...         if left[i] <= right[j]:
...             result.append(left[i])
...             i += 1
...         else:
...             result.append(right[j])
...             j += 1
...     # 把还没比较完的一边剩下的元素，直接拼到结果中
...     # 每次从 left[i] 和 right[j] 中挑一个小的加到 result，直到其中一个数组先走完（全部添加完）
...     # 注意：其中一个一定是空的，所以最多只会添加一边的剩余部分
...     result.extend(left[i:])  # 如果 left 剩下元素，就全加进来
...     result.extend(right[j:])  # 如果 right 剩下元素，就全加进来
...     return result
... 
>>> lst = [1,7,3,9,2,1,5,6]
>>> merge_sort(lst)
[1, 1, 2, 3, 5, 6, 7, 9]
```